From 1a63fdd7dbdd1cb7552bc84e06ad48ad4b889fcc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 05:28:35 -0400 Subject: [PATCH 1/4] fix(GWC-27): gate AttachInternalEventSubscriber on session readiness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AttachInternalEventSubscriber ran EnsureDistributorCreated / Register / StartPumpIfRequested with no state check, unlike AttachEventSubscriber. A premature attach would start the pump against a not-yet-Ready worker; the pump source throws SessionNotReady, PumpAsync completes every subscriber with that error and latches the distributor, and _eventDistributorStarted is never reset — so the session would reach Ready with permanently dead event streaming. Mirror AttachEventSubscriber's gate: check _state/_workerClient.State under _syncRoot and throw SessionManagerException(SessionNotReady) before the distributor is created, keeping the distributor calls outside the lock. Refs: archreview/2026-07-12/remediation/10-gateway-core.md GWC-27 --- .../Sessions/GatewaySession.cs | 27 ++++++ .../Gateway/Sessions/GatewaySessionTests.cs | 84 ++++++++++++++++++- 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index ab1e41e..0b9fa38 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -548,10 +548,37 @@ public sealed class GatewaySession /// 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. + /// + /// Gated on readiness exactly like : attaching + /// before the session and its worker are Ready throws + /// with + /// . + /// /// /// The internal subscriber's lease; dispose it to unregister. + /// + /// The session or its worker client is not Ready. + /// public IEventSubscriberLease AttachInternalEventSubscriber() { + // Readiness gate, mirroring AttachEventSubscriber (GWC-27). It must run BEFORE + // EnsureDistributorCreated: a premature attach would construct the distributor and start + // its pump against a not-yet-Ready worker, the pump source would throw SessionNotReady, + // PumpAsync would complete every subscriber with that error and latch _completed, and + // _eventDistributorStarted is never reset — so the session would reach Ready with + // permanently dead event streaming, silently, for the rest of its life. Failing loudly + // here keeps that state unreachable. The check is under _syncRoot and the distributor + // calls stay outside it, matching AttachEventSubscriber's lock discipline. + lock (_syncRoot) + { + if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready) + { + throw new SessionManagerException( + SessionManagerErrorCode.SessionNotReady, + $"Session {SessionId} is not ready for event streaming. Current state is {_state}."); + } + } + // Same sequence StartDashboardMirror uses: create the distributor (claiming the pump // 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. diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs index 7dfaef8..9e78b09 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs @@ -668,6 +668,81 @@ public sealed class GatewaySessionTests Assert.Equal(SessionState.Ready, session.State); } + /// + /// GWC-27: must refuse to + /// attach before the session is Ready. Without the gate the attach would construct and + /// start the distributor against a not-yet-Ready worker; the pump source throws + /// SessionNotReady, every subscriber is completed with that error, and the + /// distributor latches — leaving a session that reaches Ready with permanently dead + /// event streaming. The second half of the test is the load-bearing one: after the + /// failed attach the session still streams live events, proving the distributor was + /// never created or started. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor() + { + FakeWorkerClient workerClient = new(); + workerClient.Events.Add(new WorkerEvent + { + Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 1, OnDataChange = new OnDataChangeEvent() }, + }); + workerClient.Events.Add(new WorkerEvent + { + Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 2, OnDataChange = new OnDataChangeEvent() }, + }); + + // Constructed but neither worker-attached nor Ready — the premature-attach case. + await using GatewaySession session = CreateSession(); + + SessionManagerException exception = Assert.Throws( + () => session.AttachInternalEventSubscriber()); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, exception.ErrorCode); + + // Drive the session to Ready and stream: the failed attach must not have poisoned + // (or even created) the distributor, so a normal subscriber still receives events. + session.AttachWorkerClient(workerClient); + session.MarkReady(); + + using IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1); + List received = []; + using CancellationTokenSource readCts = new(TimeSpan.FromSeconds(5)); + await foreach (MxEvent mxEvent in lease.Reader.ReadAllAsync(readCts.Token)) + { + received.Add(mxEvent); + if (received.Count == 2) + { + break; + } + } + + Assert.Equal([1UL, 2UL], received.Select(mxEvent => mxEvent.WorkerSequence).ToArray()); + } + + private static GatewaySession CreateSession() + { + return new GatewaySession( + sessionId: "session-test-internal-attach", + backendName: "mxaccess", + pipeName: "mxaccess-gateway-1-session-test-internal-attach", + nonce: "nonce", + clientIdentity: "client-1", + ownerKeyId: null, + clientSessionName: "test-session", + clientCorrelationId: "client-correlation-1", + commandTimeout: TimeSpan.FromSeconds(5), + startupTimeout: TimeSpan.FromSeconds(5), + shutdownTimeout: TimeSpan.FromSeconds(5), + leaseDuration: TimeSpan.FromMinutes(30), + openedAt: DateTimeOffset.UtcNow, + eventStreaming: new SessionEventStreaming( + new MxAccessGrpcMapper(), + new EventOptions { QueueCapacity = 8 }, + NullLogger.Instance, + TimeProvider.System, + new GatewayMetrics())); + } + private static GatewaySession CreateReadySessionWithDetachGrace( IWorkerClient workerClient, TimeProvider timeProvider, @@ -855,6 +930,9 @@ public sealed class GatewaySessionTests /// Gets the count of dispose invocations. public int DisposeCount { get; private set; } + /// Events yields, in order, before completing. Empty by default. + public List Events { get; } = []; + /// public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; @@ -869,7 +947,11 @@ public sealed class GatewaySessionTests [EnumeratorCancellation] CancellationToken cancellationToken) { await Task.CompletedTask.ConfigureAwait(false); - yield break; + foreach (WorkerEvent workerEvent in Events) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return workerEvent; + } } /// From 3b6a239ed61597b8f881b9779aa948c4cd26f7b3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 05:40:33 -0400 Subject: [PATCH 2/4] fix(GWC-26): attach the alarm monitor's lease before SubscribeAlarms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunMonitorAsync issued SubscribeAlarms and the first reconcile before the internal distributor subscriber was attached (via ISessionManager .ReadAlarmEventsAsync). The pump has been running since MarkReady started the dashboard mirror and only fans to subscribers registered at fan-out time, so every transition raised in that two-round-trip window bypassed the alarm feed — and a missed Acknowledge was never repaired, because ApplyReconcile broadcast presence deltas only. - The monitor now takes the internal lease directly from its session BEFORE SubscribeAlarms and drains it after the first reconcile; window transitions buffer in the lease's bounded channel. Processing them after ApplyReconcile is order-safe (ApplyTransition handles alarms the snapshot already placed). - ISessionManager.ReadAlarmEventsAsync removed — zero remaining callers. - ApplyReconcile broadcasts an Acknowledge feed transition when a both-present alarm's state advanced to ActiveAcked. This is a feed-level repair on the AlarmFeedMessage/StreamAlarms surface rebuilt from the worker's own snapshot, not MxEvent emission, so the "never synthesize events" rule is untouched; the reasoning is recorded on ApplyReconcile. The alarm-monitor test fakes now hand the monitor a real Ready GatewaySession with a dashboard mirror, which is what makes the window reproducible. Docs: docs/Sessions.md and gateway.md alarm-monitor ordering notes. Refs: archreview/2026-07-12/remediation/10-gateway-core.md GWC-26 --- .../2026-07-12/remediation/00-tracking.md | 5 +- .../2026-07-12/remediation/10-gateway-core.md | 4 +- docs/Sessions.md | 6 +- gateway.md | 14 + .../Alarms/GatewayAlarmMonitor.cs | 38 +- .../Sessions/ISessionManager.cs | 12 - .../Sessions/SessionManager.cs | 17 - .../Alarms/AlarmFailoverEndToEndTests.cs | 76 ++- .../GatewayAlarmMonitorAttachOrderTests.cs | 508 ++++++++++++++++++ .../GatewayAlarmMonitorProviderModeTests.cs | 76 ++- .../DashboardSessionAdminServiceTests.cs | 8 - .../Gateway/Grpc/EventStreamServiceTests.cs | 8 - .../MxAccessGatewayServiceConstraintTests.cs | 8 - .../Grpc/MxAccessGatewayServiceTests.cs | 8 - .../GatewaySessionDashboardMirrorTests.cs | 5 - ...atewayGrpcAuthorizationInterceptorTests.cs | 8 - 16 files changed, 674 insertions(+), 127 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 73bdcb8..2ced6a3 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -61,8 +61,8 @@ Full design + implementation for each row lives in the linked domain doc under i |---|---|:-:|:-:|---|---|---| | GWC-24 | Medium | P1 | M | GWC-21 (coord, old tracker) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | | GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0` | -| GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed | -| GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor | +| GWC-26 | Low | P2 | M | GWC-27 | Done | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed | +| GWC-27 | Low | P2 | S | GWC-26 | Done | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor | | GWC-28 | Low | P2 | S | GWC-10 (coord, old tracker) | Not started | Gateway→worker envelope `sequence` stamped at creation, not at write | | GWC-29 | Low | — | S | — | Not started | `Invoke` deep-clones the entire request only to discard the cloned command | | GWC-30 | Info | — | S | — | Not started | Frame reader allocates a fresh 4-byte length-prefix array per frame | @@ -161,3 +161,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Operator bring-up complete: dedicated CI ed25519 key installed in windev `administrators_authorized_keys` (authorized into `dohertj2`, which owns the working MXAccess/toolchain env — a fresh OS account would break the build; the key is independently revocable), Gitea secrets `WINDEV_SSH_KEY`/`WINDEV_SSH_KNOWN_HOSTS` + variable `WINDEV_SSH_USER=dohertj2` stored, runner→`10.100.0.48:22` egress verified on the `traefik` net, issue-write confirmed. **TST-25/TST-26 → `Done`:** credentialed `windows-x86` ran GREEN on `d769244` (Gitea run #37) — Linux runner SSHed windev, checked out the SHA in `C:\build\mxaccessgw-ci` under lock, ran the x86 Worker build + `Worker.Tests`, exit 0; `nightly-windev` correctly skipped on the push event. Branch merged to `main`. Follow-ups (old tracker): revisit **TST-05** (scheduled live smoke — now covered by `nightly-windev`) and **TST-24** (client wire tests) which this unlocks. | | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | +| 2026-08-07 | **GWC-27 → `Done`, GWC-26 → `Done`** (branch `fix/gwc-26-27-alarm-attach`). GWC-27: `GatewaySession.AttachInternalEventSubscriber` now mirrors `AttachEventSubscriber`'s readiness gate under `_syncRoot`, before `EnsureDistributorCreated`, so a premature attach can no longer latch a poisoned distributor. GWC-26: the alarm monitor takes its internal lease directly from the session **before** `SubscribeAlarms` and drains it after the first reconcile; `ISessionManager.ReadAlarmEventsAsync` removed (zero remaining callers); `ApplyReconcile` now broadcasts an `Acknowledge` feed transition for a both-present alarm whose state advanced to `ActiveAcked` (feed-level repair on `AlarmFeedMessage`, not `MxEvent` synthesis). New tests `GatewaySessionTests.AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor` and `GatewayAlarmMonitorAttachOrderTests` (`TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed`, `ApplyReconcileBroadcastsAcknowledgeDelta`); the alarm-monitor fakes now hand the monitor a real Ready `GatewaySession` with a dashboard mirror so the window is actually reproducible. Verification: NonWindows build 0 warnings/0 errors; `GatewayAlarmMonitor` 16 passed, `SessionManagerTests` 38 passed, `GatewaySessionTests` 19 passed, `AlarmFailoverEndToEndTests` 2 passed. | diff --git a/archreview/2026-07-12/remediation/10-gateway-core.md b/archreview/2026-07-12/remediation/10-gateway-core.md index 7d98ccb..be879c0 100644 --- a/archreview/2026-07-12/remediation/10-gateway-core.md +++ b/archreview/2026-07-12/remediation/10-gateway-core.md @@ -10,8 +10,8 @@ This document turns the 2026-07-12 re-review's **new** Gateway Server Core findi |----|-----|------|-----|-----|--------|-------| | GWC-24 | Medium | P1 | M | GWC-21 (coord) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | | GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | -| GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed, missed Acknowledge never repaired | -| GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor permanently | +| GWC-26 | Low | P2 | M | GWC-27 | Done | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed, missed Acknowledge never repaired | +| GWC-27 | Low | P2 | S | GWC-26 | Done | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor permanently | | GWC-28 | Low | P2 | S | GWC-10 (coord) | Not started | Gateway→worker envelope `sequence` stamped at creation, not at write — non-monotonic on the wire under concurrent invokes | | GWC-29 | Low | — | S | — | Not started | `Invoke` deep-clones the entire request only to discard the cloned command | | GWC-30 | Info | — | S | — | Not started | Frame reader allocates a fresh 4-byte length-prefix array per frame | diff --git a/docs/Sessions.md b/docs/Sessions.md index 77a9db1..6d76a17 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -197,7 +197,11 @@ 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. +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 calls it directly on its session 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. + +The monitor takes that lease **before** it issues `SubscribeAlarms`, so no transition window is missed: the pump has been running since `MarkReady` started the dashboard mirror, and the distributor only fans to subscribers registered at fan-out time, so a lease taken after the subscribe + first-reconcile round trips would lose every transition raised inside that window. Transitions arriving while the monitor subscribes and reconciles buffer in the lease's bounded channel and are applied after the reconcile, which is order-safe because a live transition can update an alarm the reconciled snapshot already holds. + +`AttachInternalEventSubscriber` enforces the same readiness gate as `AttachEventSubscriber` — a session (or worker) that is not `Ready` throws `SessionNotReady` *before* the distributor is constructed. A premature attach would otherwise start the pump against a source that throws, completing every subscriber with that error and latching the distributor for the session's whole lifetime. 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). diff --git a/gateway.md b/gateway.md index 9f1e452..f978614 100644 --- a/gateway.md +++ b/gateway.md @@ -154,6 +154,20 @@ 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. +The monitor takes that internal lease **before** it sends `SubscribeAlarms`, and +drains it after the first reconcile. The pump is already running by then (the +dashboard mirror starts it at `MarkReady`) and the distributor fans only to +subscribers registered at fan-out time, so attaching after the subscribe + +reconcile round trips would drop every transition raised in that window — +including an `Acknowledge`, which the presence-only reconcile deltas would never +repair. Transitions arriving during the window buffer in the lease's bounded +channel instead. As defense in depth for any window this ordering cannot cover +(worker restart, internal-subscriber overflow disconnect), a reconcile that finds +a known alarm now reported `ActiveAcked` broadcasts an `Acknowledge` transition on +the alarm feed. That is a feed-level repair rebuilt from the worker's own +snapshot on the `StreamAlarms` surface — it is not an `MxEvent` and never reaches +`StreamEvents`, so the "never synthesize events" rule is untouched. + ### Alarm providers and failover The alarm feed has two providers, both implemented worker-side: diff --git a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs index 05714ea..d6312ba 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs @@ -210,6 +210,17 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic try { + // Attach the internal distributor subscriber BEFORE subscribing (GWC-26). The pump + // has been running since MarkReady started the dashboard mirror, and the distributor + // only fans to subscribers registered at the time of the fan-out, so a subscriber + // taken after SubscribeAlarms + the first reconcile would silently lose every + // transition raised inside that two-round-trip window — and a missed Acknowledge is + // never repaired by the presence-only reconcile deltas. Transitions arriving while we + // subscribe and reconcile simply buffer in this lease's bounded channel; if it ever + // overflowed, the internal subscriber is disconnected (it never faults the session), + // the enumeration below ends, and the supervisor loop restarts the lifecycle. + using IEventSubscriberLease alarmLease = session.AttachInternalEventSubscriber(); + await SubscribeAlarmsAsync(session.SessionId, subscription, stoppingToken).ConfigureAwait(false); await ReconcileAsync(session.SessionId, stoppingToken).ConfigureAwait(false); @@ -228,9 +239,11 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic // 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. - await foreach (MxEvent mxEvent in _sessionManager - .ReadAlarmEventsAsync(session.SessionId, linked.Token) + // mirror pump and silently lose Acknowledge/mode-change transitions. The lease was + // taken above, before SubscribeAlarms; draining it only now is order-safe because + // ApplyTransition handles alarms the reconcile already placed in the cache. + await foreach (MxEvent mxEvent in alarmLease.Reader + .ReadAllAsync(linked.Token) .ConfigureAwait(false)) { if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmTransition } @@ -511,6 +524,12 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic // Replaces the cache with the worker's authoritative snapshot, broadcasting // a synthetic transition for any alarm the live stream missed. + // + // Repair scope (GWC-26): presence deltas (Clear/Raise) plus acked-state deltas. These are + // ALARM FEED transitions (AlarmFeedMessage on the StreamAlarms/dashboard surface), rebuilt + // from the worker's own authoritative snapshot to repair what the live feed missed. They are + // not MxEvents and never reach the gRPC StreamEvents path, so this feed-level repair does not + // breach the "never synthesize events" rule, which governs MxEvent emission. private void ApplyReconcile(IEnumerable snapshots) { Dictionary next = new(StringComparer.Ordinal); @@ -536,12 +555,23 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic foreach (KeyValuePair incoming in next) { - if (!_alarms.ContainsKey(incoming.Key)) + if (!_alarms.TryGetValue(incoming.Key, out ActiveAlarmSnapshot? existing)) { Broadcast( new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Raise) }, incoming.Key); } + else if (existing.CurrentState != incoming.Value.CurrentState + && incoming.Value.CurrentState == AlarmConditionState.ActiveAcked) + { + // The alarm was already known but the worker now reports it acknowledged: the + // live Acknowledge transition never reached the feed. Without this the acked + // state is absorbed silently by the snapshot replace below and subscribers show + // the alarm unacked until it clears. + Broadcast( + new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Acknowledge) }, + incoming.Key); + } } _alarms.Clear(); diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs index c9431a1..f9c200e 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs @@ -43,18 +43,6 @@ 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 428d99f..23abece 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs @@ -1,5 +1,4 @@ using System.Diagnostics.CodeAnalysis; -using System.Runtime.CompilerServices; using System.Security.Cryptography; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging; @@ -191,22 +190,6 @@ 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, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs index e927439..5ed3994 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs @@ -8,6 +8,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms; using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Sessions; +using ZB.MOM.WW.MxGateway.Server.Workers; namespace ZB.MOM.WW.MxGateway.Tests.Alarms; @@ -420,6 +421,9 @@ public sealed class AlarmFailoverEndToEndTests string? ownerKeyId, CancellationToken cancellationToken) { + // The monitor attaches its internal subscriber directly on this session, so the + // session has to be a genuinely Ready one with a worker client feeding the + // distributor pump — EmitEvent writes into that worker's event stream. GatewaySession session = new( Guid.NewGuid().ToString("N"), "Galaxy", @@ -432,6 +436,8 @@ public sealed class AlarmFailoverEndToEndTests TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow); + session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader)); + session.MarkReady(); return Task.FromResult(session); } @@ -460,29 +466,9 @@ public sealed class AlarmFailoverEndToEndTests } /// - public async IAsyncEnumerable ReadEventsAsync( + public IAsyncEnumerable ReadEventsAsync( string sessionId, - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken)) - { - yield return workerEvent; - } - } - - /// - 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; - } - } - } + CancellationToken cancellationToken) => throw new NotSupportedException(); /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) @@ -509,4 +495,50 @@ public sealed class AlarmFailoverEndToEndTests /// public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; } + + /// Ready worker client whose event stream is the fake session manager's channel. + private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient + { + /// + public string SessionId { get; } = sessionId; + + /// + public int? ProcessId { get; } = 4321; + + /// + public WorkerClientState State { get; } = WorkerClientState.Ready; + + /// + public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task InvokeAsync( + WorkerCommand command, + TimeSpan timeout, + CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); + + /// + public async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return workerEvent; + } + } + + /// + public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public void Kill(string reason) + { + } + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs new file mode 100644 index 0000000..1b99cc2 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs @@ -0,0 +1,508 @@ +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Alarms; +using ZB.MOM.WW.MxGateway.Server.Configuration; +using ZB.MOM.WW.MxGateway.Server.Grpc; +using ZB.MOM.WW.MxGateway.Server.Metrics; +using ZB.MOM.WW.MxGateway.Server.Sessions; +using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; + +namespace ZB.MOM.WW.MxGateway.Tests.Alarms; + +/// +/// GWC-26 regression tests for the alarm monitor's startup ordering. Unlike the +/// sibling alarm-monitor tests, the session manager here hands the monitor a REAL +/// that is driven to Ready with a dashboard mirror, so the +/// distributor pump is already running when the monitor attaches — the production +/// condition under which a late internal subscriber silently misses everything the pump +/// already fanned. +/// +public sealed class GatewayAlarmMonitorAttachOrderTests +{ + private const string AlarmReference = "Galaxy!Area.Tank01.Hi"; + private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30); + + /// + /// The monitor must take its internal distributor lease BEFORE issuing + /// SubscribeAlarms, so transitions the worker emits during the + /// subscribe + first-reconcile window buffer in the lease instead of being fanned to a + /// subscriber set the monitor has not joined yet. The test parks the monitor inside its + /// SubscribeAlarms round trip, emits a Raise and an Acknowledge, waits until the + /// dashboard mirror proves the pump has already fanned both, and only then releases the + /// monitor: with a late attach both transitions are lost to the alarm feed forever. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed() + { + using GatewayMetrics metrics = new(); + await using FakeSessionManager sessions = new(); + sessions.HoldSubscribeUntilReleased(); + using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics); + + using CancellationTokenSource cts = new(); + await monitor.StartAsync(cts.Token); + await sessions.WaitForSubscribeStartAsync(WaitTimeout); + + // A live feed subscriber, drained past its baseline ProviderStatus so it is registered + // before any window transition is broadcast. + List received = []; + TaskCompletionSource baselineReceived = new(TaskCreationOptions.RunContinuationsAsynchronously); + using CancellationTokenSource streamCts = new(); + Task reader = ReadFeedAsync(monitor, received, baselineReceived, streamCts.Token); + await baselineReceived.Task.WaitAsync(WaitTimeout); + + // The window: the worker reports a raise and an acknowledge while the monitor is still + // waiting for its SubscribeAlarms reply. + sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise)); + sessions.EmitEvent(Transition(2, AlarmTransitionKind.Acknowledge)); + + // The dashboard mirror is an independent distributor subscriber: once it has both + // events, the pump has provably already fanned them, so a subscriber that registers + // after this point can never receive them. + await WaitUntilAsync(() => sessions.Broadcaster.Captures.Count == 2, WaitTimeout); + + sessions.ReleaseSubscribe(); + + AlarmFeedMessage raise = await WaitForAsync( + received, + m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition + && m.Transition.TransitionKind == AlarmTransitionKind.Raise, + WaitTimeout); + AlarmFeedMessage acknowledge = await WaitForAsync( + received, + m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition + && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge, + WaitTimeout); + + Assert.Equal(AlarmReference, raise.Transition.AlarmFullReference); + Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference); + + await streamCts.CancelAsync(); + await reader; + await cts.CancelAsync(); + await monitor.StopAsync(CancellationToken.None); + } + + /// + /// Defense in depth for any window the attach reorder cannot cover (worker restart, + /// internal-subscriber overflow disconnect): when a reconcile snapshot reports an alarm + /// the cache already holds but with the state advanced to + /// , the monitor broadcasts an + /// feed transition. Before the fix the + /// acked state was absorbed silently by the snapshot replace, leaving live subscribers + /// showing the alarm unacked until it cleared. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ApplyReconcileBroadcastsAcknowledgeDelta() + { + using GatewayMetrics metrics = new(); + await using FakeSessionManager sessions = new(); + using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics); + + using CancellationTokenSource cts = new(); + await monitor.StartAsync(cts.Token); + await sessions.WaitForSubscribeStartAsync(WaitTimeout); + + // Seed the cache with an active (unacked) alarm and keep the reconcile snapshot in + // agreement, so a periodic reconcile pass cannot clear it out from under the test. + sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active)); + sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise)); + await WaitUntilAsync( + () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference), + WaitTimeout); + + // Subscribe AFTER the raise: this reader's snapshot carries the alarm, so every + // transition it observes from here on is a reconcile-derived broadcast. + List received = []; + TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously); + using CancellationTokenSource streamCts = new(); + Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true); + await snapshotComplete.Task.WaitAsync(WaitTimeout); + + // The worker now reports the same alarm acknowledged. A provider-mode event forces an + // immediate reconcile pass so the test does not wait on the periodic timer. + sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.ActiveAcked)); + sessions.EmitEvent(new MxEvent + { + Family = MxEventFamily.OnAlarmProviderModeChanged, + WorkerSequence = 2, + OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent + { + Mode = AlarmProviderMode.Alarmmgr, + Reason = "probe", + At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + }); + + AlarmFeedMessage acknowledge = await WaitForAsync( + received, + m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition + && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge, + WaitTimeout); + Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference); + + lock (received) + { + AlarmFeedMessage[] transitions = received + .Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition) + .ToArray(); + AlarmFeedMessage single = Assert.Single(transitions); + Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind); + } + + await streamCts.CancelAsync(); + await reader; + await cts.CancelAsync(); + await monitor.StopAsync(CancellationToken.None); + } + + private static GatewayAlarmMonitor CreateMonitor(FakeSessionManager sessions, GatewayMetrics metrics) + { + AlarmsOptions options = new() + { + Enabled = true, + SubscriptionExpression = @"\\NODE\Galaxy!Area", + }; + return new GatewayAlarmMonitor( + sessions, + new StubWatchListResolver(), + metrics, + Microsoft.Extensions.Options.Options.Create(new GatewayOptions { Alarms = options }), + NullLogger.Instance); + } + + // Drains the monitor's feed into received, signalling gate on the first message (the + // baseline ProviderStatus) or, when untilSnapshotComplete is set, on SnapshotComplete. + private static Task ReadFeedAsync( + GatewayAlarmMonitor monitor, + List received, + TaskCompletionSource gate, + CancellationToken cancellationToken, + bool untilSnapshotComplete = false) + { + return Task.Run( + async () => + { + try + { + await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, cancellationToken)) + { + bool opensGate = !untilSnapshotComplete + || message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete; + + // Record only what arrives AFTER the gate opened: everything up to and + // including the gate message is this subscriber's snapshot preamble, not + // a live broadcast, so assertions stay about broadcasts alone. + lock (received) + { + if (gate.Task.IsCompleted) + { + received.Add(message); + } + } + + if (opensGate) + { + gate.TrySetResult(); + } + } + } + catch (OperationCanceledException) + { + // Expected when the test cancels the stream. + } + }, + CancellationToken.None); + } + + private static MxEvent Transition(ulong sequence, AlarmTransitionKind kind) => new() + { + Family = MxEventFamily.OnAlarmTransition, + WorkerSequence = sequence, + OnAlarmTransition = new OnAlarmTransitionEvent + { + AlarmFullReference = AlarmReference, + SourceObjectReference = "Tank01", + AlarmTypeName = "AnalogLimitAlarm.Hi", + TransitionKind = kind, + Severity = 500, + SourceProvider = AlarmProviderMode.Alarmmgr, + TransitionTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + }; + + private static ActiveAlarmSnapshot Snapshot(AlarmConditionState state) => new() + { + AlarmFullReference = AlarmReference, + SourceObjectReference = "Tank01", + AlarmTypeName = "AnalogLimitAlarm.Hi", + CurrentState = state, + Severity = 500, + SourceProvider = AlarmProviderMode.Alarmmgr, + }; + + private static async Task WaitForAsync( + List received, + Func predicate, + TimeSpan timeout) + { + DateTime deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + lock (received) + { + AlarmFeedMessage? match = received.FirstOrDefault(predicate); + if (match is not null) + { + return match; + } + } + + await Task.Delay(25); + } + + throw new TimeoutException("No matching AlarmFeedMessage was received in time."); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + DateTime deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return; + } + + await Task.Delay(25); + } + + throw new TimeoutException("Condition was not met in time."); + } + + /// that resolves an empty watch-list. + private sealed class StubWatchListResolver : IAlarmWatchListResolver + { + /// + public Task> ResolveAsync( + AlarmsOptions options, + CancellationToken cancellationToken = default) => + Task.FromResult>([]); + } + + /// + /// Session manager that hands the monitor a real driven to + /// Ready with a dashboard mirror, so the distributor pump is running before the monitor + /// attaches. pushes worker events through the fake worker client + /// into that pump, exactly as a live worker would. + /// + private sealed class FakeSessionManager : ISessionManager, IAsyncDisposable + { + private readonly Channel _events = Channel.CreateUnbounded(); + private readonly TaskCompletionSource _subscribeStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly object _sync = new(); + private TaskCompletionSource _subscribeGate = CreateReleasedGate(); + private ActiveAlarmSnapshot[] _reconcileSnapshot = []; + private GatewaySession? _session; + + /// Dashboard mirror attached to the session; proves what the pump has fanned. + public RecordingDashboardEventBroadcaster Broadcaster { get; } = new(); + + /// Re-arms the gate so SubscribeAlarms parks until . + public void HoldSubscribeUntilReleased() => + _subscribeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + /// Releases a gate armed by . + public void ReleaseSubscribe() => _subscribeGate.TrySetResult(); + + /// Completes once the monitor's SubscribeAlarms command has arrived. + /// The maximum time to wait. + /// A task that completes when the command arrives. + public Task WaitForSubscribeStartAsync(TimeSpan timeout) => _subscribeStarted.Task.WaitAsync(timeout); + + /// Sets the active-alarm snapshot every QueryActiveAlarms reconcile returns. + /// The snapshots to report. + public void SetReconcileSnapshot(params ActiveAlarmSnapshot[] snapshots) + { + lock (_sync) + { + _reconcileSnapshot = snapshots; + } + } + + /// Pushes a worker event into the session's distributor pump. + /// The event to push. + public void EmitEvent(MxEvent mxEvent) => + _events.Writer.TryWrite(new WorkerEvent { Event = mxEvent }); + + /// + public Task OpenSessionAsync( + SessionOpenRequest request, + string? clientIdentity, + string? ownerKeyId, + CancellationToken cancellationToken) + { + GatewaySession session = new( + sessionId: "session-alarm-attach-order", + backendName: "Galaxy", + pipeName: "mxaccess-gateway-1-session-alarm-attach-order", + nonce: "nonce", + clientIdentity: clientIdentity, + ownerKeyId: ownerKeyId, + clientSessionName: request.ClientSessionName, + clientCorrelationId: request.ClientCorrelationId, + commandTimeout: TimeSpan.FromSeconds(30), + startupTimeout: TimeSpan.FromSeconds(30), + shutdownTimeout: TimeSpan.FromSeconds(30), + leaseDuration: TimeSpan.FromMinutes(30), + openedAt: DateTimeOffset.UtcNow, + eventStreaming: new SessionEventStreaming( + new MxAccessGrpcMapper(), + new EventOptions { QueueCapacity = 64 }, + NullLogger.Instance, + TimeProvider.System, + new GatewayMetrics(), + Broadcaster)); + session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader)); + + // MarkReady starts the dashboard mirror, and with it the distributor pump — the + // production precondition this regression depends on. + session.MarkReady(); + _session = session; + return Task.FromResult(session); + } + + /// + public async Task InvokeAsync( + string sessionId, + WorkerCommand command, + CancellationToken cancellationToken) + { + MxCommandReply reply = new() + { + ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok }, + }; + + switch (command.Command?.Kind) + { + case MxCommandKind.SubscribeAlarms: + _subscribeStarted.TrySetResult(); + await _subscribeGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + break; + case MxCommandKind.QueryActiveAlarms: + QueryActiveAlarmsReplyPayload payload = new(); + lock (_sync) + { + payload.Snapshots.AddRange(_reconcileSnapshot.Select(snapshot => snapshot.Clone())); + } + + reply.QueryActiveAlarms = payload; + break; + } + + return new WorkerCommandReply { Reply = reply }; + } + + /// + public IAsyncEnumerable ReadEventsAsync( + string sessionId, + CancellationToken cancellationToken) => throw new NotSupportedException(); + + /// + public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) + { + session = _session; + return session is not null; + } + + /// + public Task CloseSessionAsync(string sessionId, CancellationToken cancellationToken) + { + _events.Writer.TryComplete(); + return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false)); + } + + /// + public Task KillWorkerAsync(string sessionId, string reason, CancellationToken cancellationToken) => + Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false)); + + /// + public Task CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) => + Task.FromResult(0); + + /// + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// Disposes the session the fake handed out. + /// A task that represents the asynchronous operation. + public async ValueTask DisposeAsync() + { + _events.Writer.TryComplete(); + if (_session is not null) + { + await _session.DisposeAsync().ConfigureAwait(false); + } + } + + private static TaskCompletionSource CreateReleasedGate() + { + TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + gate.SetResult(); + return gate; + } + } + + /// Worker client whose event stream is a test-driven channel. + private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient + { + /// + public string SessionId { get; } = sessionId; + + /// + public int? ProcessId { get; } = 4321; + + /// + public WorkerClientState State { get; } = WorkerClientState.Ready; + + /// + public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task InvokeAsync( + WorkerCommand command, + TimeSpan timeout, + CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); + + /// + public async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return workerEvent; + } + } + + /// + public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public void Kill(string reason) + { + } + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs index 63944d5..13eaa66 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs @@ -10,6 +10,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms; using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Sessions; +using ZB.MOM.WW.MxGateway.Server.Workers; namespace ZB.MOM.WW.MxGateway.Tests.Alarms; @@ -733,6 +734,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests string? ownerKeyId, CancellationToken cancellationToken) { + // The monitor attaches its internal subscriber directly on this session, so the + // session has to be a genuinely Ready one with a worker client feeding the + // distributor pump — EmitEvent writes into that worker's event stream. GatewaySession session = new( Guid.NewGuid().ToString("N"), "Galaxy", @@ -745,6 +749,8 @@ public sealed class GatewayAlarmMonitorProviderModeTests TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow); + session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader)); + session.MarkReady(); return Task.FromResult(session); } @@ -773,29 +779,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests } /// - public async IAsyncEnumerable ReadEventsAsync( + public IAsyncEnumerable ReadEventsAsync( string sessionId, - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken)) - { - yield return workerEvent; - } - } - - /// - 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; - } - } - } + CancellationToken cancellationToken) => throw new NotSupportedException(); /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) @@ -822,4 +808,50 @@ public sealed class GatewayAlarmMonitorProviderModeTests /// public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; } + + /// Ready worker client whose event stream is the fake session manager's channel. + private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient + { + /// + public string SessionId { get; } = sessionId; + + /// + public int? ProcessId { get; } = 4321; + + /// + public WorkerClientState State { get; } = WorkerClientState.Ready; + + /// + public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task InvokeAsync( + WorkerCommand command, + TimeSpan timeout, + CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); + + /// + public async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return workerEvent; + } + } + + /// + public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public void Kill(string reason) + { + } + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } } 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 611447b..b0cdf77 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs @@ -378,14 +378,6 @@ 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 4943e8d..f871fe9 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs @@ -755,14 +755,6 @@ 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 71b0ae2..885bffb 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs @@ -948,14 +948,6 @@ public sealed class MxAccessGatewayServiceConstraintTests } } - /// - 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/MxAccessGatewayServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs index a26eeb7..5ad05a8 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs @@ -616,14 +616,6 @@ public sealed class MxAccessGatewayServiceTests } } - /// - 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/GatewaySessionDashboardMirrorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs index 3b12acf..b103a29 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs @@ -352,11 +352,6 @@ 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/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs index 189b8fa..618612f 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs @@ -579,14 +579,6 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests return AsyncEnumerable.Empty(); } - /// - public IAsyncEnumerable ReadAlarmEventsAsync( - string sessionId, - CancellationToken cancellationToken) - { - return AsyncEnumerable.Empty(); - } - /// public Task CloseSessionAsync( string sessionId, From 1a75f61ebe52540aceb5c8452fb8864b57196ed8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 05:49:34 -0400 Subject: [PATCH 3/4] test(GWC-26): deflake ApplyReconcileBroadcastsAcknowledgeDelta MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Seeding the cache with a live Raise transition raced the first reconcile: when the reconcile snapshot populated the cache first, the still-buffered live Raise was applied — and broadcast — after the test's feed subscriber had registered, so the exactly-one-transition assertion saw two. Seed through a reconcile pass instead (forced by a provider-mode probe, as the acked step already did) so no live transition is ever in flight. Verified: 5 consecutive full alarm-monitor runs green, and the test still fails (timeout) with the ApplyReconcile acked-delta branch removed. --- .../GatewayAlarmMonitorAttachOrderTests.cs | 40 +++++++++++-------- 1 file changed, 24 insertions(+), 16 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs index 1b99cc2..48ad14d 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs @@ -110,15 +110,19 @@ public sealed class GatewayAlarmMonitorAttachOrderTests await monitor.StartAsync(cts.Token); await sessions.WaitForSubscribeStartAsync(WaitTimeout); - // Seed the cache with an active (unacked) alarm and keep the reconcile snapshot in - // agreement, so a periodic reconcile pass cannot clear it out from under the test. + // Seed the cache with an active (unacked) alarm through a reconcile rather than a live + // transition: a buffered live transition could still be in flight when the cache first + // shows the alarm, and would then broadcast to the reader below and pollute the + // exactly-one-transition assertion. A provider-mode event forces the reconcile + // immediately, so the test never waits on the periodic timer. sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active)); - sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise)); + sessions.EmitEvent(ProviderModeProbe(1)); await WaitUntilAsync( - () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference), + () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference + && alarm.CurrentState == AlarmConditionState.Active), WaitTimeout); - // Subscribe AFTER the raise: this reader's snapshot carries the alarm, so every + // Subscribe AFTER the seed: this reader's snapshot carries the alarm, so every // transition it observes from here on is a reconcile-derived broadcast. List received = []; TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -129,17 +133,7 @@ public sealed class GatewayAlarmMonitorAttachOrderTests // The worker now reports the same alarm acknowledged. A provider-mode event forces an // immediate reconcile pass so the test does not wait on the periodic timer. sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.ActiveAcked)); - sessions.EmitEvent(new MxEvent - { - Family = MxEventFamily.OnAlarmProviderModeChanged, - WorkerSequence = 2, - OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent - { - Mode = AlarmProviderMode.Alarmmgr, - Reason = "probe", - At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), - }, - }); + sessions.EmitEvent(ProviderModeProbe(2)); AlarmFeedMessage acknowledge = await WaitForAsync( received, @@ -238,6 +232,20 @@ public sealed class GatewayAlarmMonitorAttachOrderTests }, }; + // A no-op provider-mode event. The monitor forces an immediate reconcile after every one, + // which is how these tests drive a reconcile pass without waiting on the periodic timer. + private static MxEvent ProviderModeProbe(ulong sequence) => new() + { + Family = MxEventFamily.OnAlarmProviderModeChanged, + WorkerSequence = sequence, + OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent + { + Mode = AlarmProviderMode.Alarmmgr, + Reason = "probe", + At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + }; + private static ActiveAlarmSnapshot Snapshot(AlarmConditionState state) => new() { AlarmFullReference = AlarmReference, From 09ccd9561f0e1ec3a47e664bb263d9592a93ab89 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:00:16 -0400 Subject: [PATCH 4/4] docs(GWC-26): record alarm feed repairs as at-least-once; share the channel worker fake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code-review follow-up on fix/gwc-26-27-alarm-attach. ApplyReconcile's snapshot-derived feed repairs are at-least-once, not exactly-once: a reconcile reads the worker's current state while the matching live transition may still be buffered in the monitor's lease, so both broadcast and the duplicates are indistinguishable on the alarm feed. This pre-dates the acked-state delta — the Raise/Clear presence repair has always had it, since nothing serializes a reconcile pass against the in-flight live stream — so closing it (serialization or timestamp dedup) stays out of scope for a P2 fix. Documented instead, with the consumer contract stated explicitly (apply transitions idempotently, never as an increment or toggle): - ApplyReconcile gains a "Delivery semantics" comment. - gateway.md softens the "defense in depth" prose to state the semantics. - docs/Sessions.md carries the same caveat on the alarm-feed description. - Tracker change-log records it as a known pre-existing characteristic and a candidate finding for the next review cycle. Also hoists the ChannelWorkerClient fake — duplicated across the three alarm test files — into TestSupport/, dropping the usings it took with it. --- .../2026-07-12/remediation/00-tracking.md | 1 + docs/Sessions.md | 2 + gateway.md | 19 ++++-- .../Alarms/GatewayAlarmMonitor.cs | 8 +++ .../Alarms/AlarmFailoverEndToEndTests.cs | 49 +------------- .../GatewayAlarmMonitorAttachOrderTests.cs | 48 -------------- .../GatewayAlarmMonitorProviderModeTests.cs | 49 +------------- .../TestSupport/ChannelWorkerClient.cs | 66 +++++++++++++++++++ 8 files changed, 93 insertions(+), 149 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/TestSupport/ChannelWorkerClient.cs diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 2ced6a3..c0af6f6 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -162,3 +162,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | | 2026-08-07 | **GWC-27 → `Done`, GWC-26 → `Done`** (branch `fix/gwc-26-27-alarm-attach`). GWC-27: `GatewaySession.AttachInternalEventSubscriber` now mirrors `AttachEventSubscriber`'s readiness gate under `_syncRoot`, before `EnsureDistributorCreated`, so a premature attach can no longer latch a poisoned distributor. GWC-26: the alarm monitor takes its internal lease directly from the session **before** `SubscribeAlarms` and drains it after the first reconcile; `ISessionManager.ReadAlarmEventsAsync` removed (zero remaining callers); `ApplyReconcile` now broadcasts an `Acknowledge` feed transition for a both-present alarm whose state advanced to `ActiveAcked` (feed-level repair on `AlarmFeedMessage`, not `MxEvent` synthesis). New tests `GatewaySessionTests.AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor` and `GatewayAlarmMonitorAttachOrderTests` (`TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed`, `ApplyReconcileBroadcastsAcknowledgeDelta`); the alarm-monitor fakes now hand the monitor a real Ready `GatewaySession` with a dashboard mirror so the window is actually reproducible. Verification: NonWindows build 0 warnings/0 errors; `GatewayAlarmMonitor` 16 passed, `SessionManagerTests` 38 passed, `GatewaySessionTests` 19 passed, `AlarmFailoverEndToEndTests` 2 passed. | +| 2026-08-07 | Code review of `fix/gwc-26-27-alarm-attach` surfaced a **known pre-existing characteristic, now documented**: the alarm monitor's reconcile-derived feed repairs are **at-least-once, not exactly-once**. A reconcile reads the worker's current state while the matching live transition may still be buffered in the monitor's internal lease, so both broadcast and the duplicates are indistinguishable on the alarm feed (`StreamAlarms` + dashboard alarm hub). This pre-dates GWC-26 — the Raise/Clear presence repair has always had it, since nothing serializes a reconcile pass against the in-flight live stream — so closing it (reconcile/live serialization or transition-timestamp dedup) was ruled out of scope for a P2 fix. Documented instead in `GatewayAlarmMonitor.ApplyReconcile`, `gateway.md`, and `docs/Sessions.md`, with the consumer-side contract stated explicitly (apply transitions idempotently — "set this alarm to this state", never increment/toggle). **Candidate finding for the next review cycle.** | diff --git a/docs/Sessions.md b/docs/Sessions.md index 6d76a17..65e26fe 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -201,6 +201,8 @@ The single worker event channel has exactly one direct reader: the `SessionEvent The monitor takes that lease **before** it issues `SubscribeAlarms`, so no transition window is missed: the pump has been running since `MarkReady` started the dashboard mirror, and the distributor only fans to subscribers registered at fan-out time, so a lease taken after the subscribe + first-reconcile round trips would lose every transition raised inside that window. Transitions arriving while the monitor subscribes and reconciles buffer in the lease's bounded channel and are applied after the reconcile, which is order-safe because a live transition can update an alarm the reconciled snapshot already holds. +The repair transitions the monitor's reconcile broadcasts on the alarm feed (Raise/Clear presence deltas and the acked-state delta) are **at-least-once**: a reconcile reads the worker's current state while the matching live transition may still be buffered in the lease, so both can be broadcast and the duplicates are indistinguishable — alarm-feed consumers must apply transitions idempotently. This is a property of the reconcile design, not of the buffering above. + `AttachInternalEventSubscriber` enforces the same readiness gate as `AttachEventSubscriber` — a session (or worker) that is not `Ready` throws `SessionNotReady` *before* the distributor is constructed. A premature attach would otherwise start the pump against a source that throws, completing every subscriber with that error and latching the distributor for the session's whole lifetime. 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). diff --git a/gateway.md b/gateway.md index f978614..1c15cd1 100644 --- a/gateway.md +++ b/gateway.md @@ -161,13 +161,22 @@ subscribers registered at fan-out time, so attaching after the subscribe + reconcile round trips would drop every transition raised in that window — including an `Acknowledge`, which the presence-only reconcile deltas would never repair. Transitions arriving during the window buffer in the lease's bounded -channel instead. As defense in depth for any window this ordering cannot cover -(worker restart, internal-subscriber overflow disconnect), a reconcile that finds -a known alarm now reported `ActiveAcked` broadcasts an `Acknowledge` transition on -the alarm feed. That is a feed-level repair rebuilt from the worker's own -snapshot on the `StreamAlarms` surface — it is not an `MxEvent` and never reaches +channel instead. As a backstop for any window this ordering cannot cover (worker +restart, internal-subscriber overflow disconnect), a reconcile that finds a known +alarm now reported `ActiveAcked` broadcasts an `Acknowledge` transition on the +alarm feed. That is a feed-level repair rebuilt from the worker's own snapshot on +the `StreamAlarms` surface — it is not an `MxEvent` and never reaches `StreamEvents`, so the "never synthesize events" rule is untouched. +**Feed repair transitions are at-least-once, not exactly-once.** A reconcile reads +the worker's current state while the corresponding live transition may still be +buffered in the monitor's lease, so both can broadcast and the two are +indistinguishable on the feed. This applies to the acked-state delta and equally +to the older Raise/Clear presence repair: nothing serializes a reconcile pass +against the in-flight live stream. Alarm-feed consumers (`StreamAlarms` clients +and the dashboard alarm hub) must apply transitions idempotently — treat one as +"set this alarm to this state", never as an increment or a toggle. + ### Alarm providers and failover The alarm feed has two providers, both implemented worker-side: diff --git a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs index d6312ba..0f1e1cc 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs @@ -530,6 +530,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic // from the worker's own authoritative snapshot to repair what the live feed missed. They are // not MxEvents and never reach the gRPC StreamEvents path, so this feed-level repair does not // breach the "never synthesize events" rule, which governs MxEvent emission. + // + // Delivery semantics: feed repair transitions are AT-LEAST-ONCE, not exactly-once. A reconcile + // reads the worker's current state while the corresponding live transition may still be + // buffered in the alarm lease's channel; both then broadcast, and the two are indistinguishable + // on the feed. This is inherent to the reconcile design and pre-dates the acked-state delta + // (the Raise/Clear repair has always had it), since nothing serializes a reconcile against the + // in-flight live stream. Consumers must therefore treat alarm state idempotently — apply a + // transition as "set the alarm to this state", never as an increment or a toggle. private void ApplyReconcile(IEnumerable snapshots) { Dictionary next = new(StringComparer.Ordinal); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs index 5ed3994..dd0813e 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs @@ -1,5 +1,4 @@ using System.Diagnostics.CodeAnalysis; -using System.Runtime.CompilerServices; using System.Threading.Channels; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging.Abstractions; @@ -8,7 +7,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms; using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Sessions; -using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; namespace ZB.MOM.WW.MxGateway.Tests.Alarms; @@ -495,50 +494,4 @@ public sealed class AlarmFailoverEndToEndTests /// public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; } - - /// Ready worker client whose event stream is the fake session manager's channel. - private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient - { - /// - public string SessionId { get; } = sessionId; - - /// - public int? ProcessId { get; } = 4321; - - /// - public WorkerClientState State { get; } = WorkerClientState.Ready; - - /// - public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; - - /// - public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public Task InvokeAsync( - WorkerCommand command, - TimeSpan timeout, - CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); - - /// - public async IAsyncEnumerable ReadEventsAsync( - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return workerEvent; - } - } - - /// - public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public void Kill(string reason) - { - } - - /// - public ValueTask DisposeAsync() => ValueTask.CompletedTask; - } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs index 48ad14d..061931b 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs @@ -1,5 +1,4 @@ using System.Diagnostics.CodeAnalysis; -using System.Runtime.CompilerServices; using System.Threading.Channels; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging.Abstractions; @@ -9,7 +8,6 @@ using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Grpc; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Sessions; -using ZB.MOM.WW.MxGateway.Server.Workers; using ZB.MOM.WW.MxGateway.Tests.TestSupport; namespace ZB.MOM.WW.MxGateway.Tests.Alarms; @@ -467,50 +465,4 @@ public sealed class GatewayAlarmMonitorAttachOrderTests return gate; } } - - /// Worker client whose event stream is a test-driven channel. - private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient - { - /// - public string SessionId { get; } = sessionId; - - /// - public int? ProcessId { get; } = 4321; - - /// - public WorkerClientState State { get; } = WorkerClientState.Ready; - - /// - public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; - - /// - public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public Task InvokeAsync( - WorkerCommand command, - TimeSpan timeout, - CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); - - /// - public async IAsyncEnumerable ReadEventsAsync( - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return workerEvent; - } - } - - /// - public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public void Kill(string reason) - { - } - - /// - public ValueTask DisposeAsync() => ValueTask.CompletedTask; - } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs index 13eaa66..eef1fa3 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs @@ -1,6 +1,5 @@ using System.Diagnostics.CodeAnalysis; using System.Diagnostics.Metrics; -using System.Runtime.CompilerServices; using System.Threading.Channels; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging.Abstractions; @@ -10,7 +9,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms; using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Sessions; -using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; namespace ZB.MOM.WW.MxGateway.Tests.Alarms; @@ -808,50 +807,4 @@ public sealed class GatewayAlarmMonitorProviderModeTests /// public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; } - - /// Ready worker client whose event stream is the fake session manager's channel. - private sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient - { - /// - public string SessionId { get; } = sessionId; - - /// - public int? ProcessId { get; } = 4321; - - /// - public WorkerClientState State { get; } = WorkerClientState.Ready; - - /// - public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; - - /// - public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public Task InvokeAsync( - WorkerCommand command, - TimeSpan timeout, - CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); - - /// - public async IAsyncEnumerable ReadEventsAsync( - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return workerEvent; - } - } - - /// - public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; - - /// - public void Kill(string reason) - { - } - - /// - public ValueTask DisposeAsync() => ValueTask.CompletedTask; - } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/ChannelWorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/ChannelWorkerClient.cs new file mode 100644 index 0000000..378d125 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/ChannelWorkerClient.cs @@ -0,0 +1,66 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Workers; + +namespace ZB.MOM.WW.MxGateway.Tests.TestSupport; + +/// +/// Always- whose event +/// stream is a channel the test writes to. Lets a test attach a real +/// GatewaySession to a worker it drives by hand — the session can be marked Ready, +/// its SessionEventDistributor pump then drains this channel, and the test controls +/// exactly when each event is fanned. +/// +/// +/// Commands are not scripted here: returns an empty reply, because +/// the consumers of this fake route commands through their own ISessionManager double +/// rather than through the worker client. Use a purpose-built worker client instead when a +/// test needs command behavior. +/// +/// Session identifier the client reports. +/// Channel whose events yields, in order. +public sealed class ChannelWorkerClient(string sessionId, ChannelReader events) : IWorkerClient +{ + /// + public string SessionId { get; } = sessionId; + + /// + public int? ProcessId { get; } = 4321; + + /// + public WorkerClientState State { get; } = WorkerClientState.Ready; + + /// + public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task InvokeAsync( + WorkerCommand command, + TimeSpan timeout, + CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); + + /// + public async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return workerEvent; + } + } + + /// + public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public void Kill(string reason) + { + } + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; +}