From 1eb00276bc7a4d158baba0f1ebc9973031d1fec4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:45 -0400 Subject: [PATCH 1/5] fix(CLI-01): Go Events() emits terminal ErrSlowConsumer on overflow Events()/EventsAfter() previously closed the channel silently when the 16-slot buffer overflowed, indistinguishable from a graceful server end. Reserve one slot and emit a terminal EventResult wrapping the new exported ErrSlowConsumer so overflow is always observable. The blocking SubscribeEvents path (gRPC flow-controlled) is unchanged. archreview: CLI-01 (P0). Verified: gofmt clean, go build ./..., go test ./... (+ -race) pass. --- clients/go/mxgateway/client_session_test.go | 53 +++++++++++++++++++++ clients/go/mxgateway/errors.go | 8 ++++ clients/go/mxgateway/session.go | 44 +++++++++++++++-- 3 files changed, 101 insertions(+), 4 deletions(-) diff --git a/clients/go/mxgateway/client_session_test.go b/clients/go/mxgateway/client_session_test.go index bc5d597..fa1f85b 100644 --- a/clients/go/mxgateway/client_session_test.go +++ b/clients/go/mxgateway/client_session_test.go @@ -147,6 +147,59 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing. } } +func TestEventsSlowConsumerYieldsErrSlowConsumerBeforeClose(t *testing.T) { + fake := &fakeGatewayServer{ + streamStarted: make(chan struct{}), + streamDone: make(chan struct{}), + streamEventCount: 64, + } + client, cleanup := newBufconnClient(t, fake) + defer cleanup() + session := NewSessionForID(client, "session-1") + + events, err := session.EventsAfter(context.Background(), 0) + if err != nil { + t.Fatalf("EventsAfter() error = %v", err) + } + <-fake.streamStarted + + // Do not drain the channel so the buffer overflows. The stream stops once + // the slow-consumer cancel fires on the producer side. + select { + case <-fake.streamDone: + case <-time.After(2 * time.Second): + t.Fatal("event stream did not stop after buffer overflow") + } + + var last EventResult + gotResult := false + for { + select { + case res, ok := <-events: + if !ok { + if !gotResult { + t.Fatal("events channel closed without yielding any result") + } + if !errors.Is(last.Err, ErrSlowConsumer) { + t.Fatalf("final event result err = %v, want one wrapping ErrSlowConsumer", last.Err) + } + var gwErr *GatewayError + if !errors.As(last.Err, &gwErr) { + t.Fatalf("final event result err is %T, want *GatewayError", last.Err) + } + if gwErr.Op != "stream events" { + t.Fatalf("final event result err Op = %q, want %q", gwErr.Op, "stream events") + } + return + } + last = res + gotResult = true + case <-time.After(2 * time.Second): + t.Fatal("events channel did not close after slow-consumer termination") + } + } +} + func TestSessionHelpersBuildCommandsAndExposeRawReply(t *testing.T) { fake := &fakeGatewayServer{ invokeReply: &pb.MxCommandReply{ diff --git a/clients/go/mxgateway/errors.go b/clients/go/mxgateway/errors.go index 92dd486..778e56e 100644 --- a/clients/go/mxgateway/errors.go +++ b/clients/go/mxgateway/errors.go @@ -1,11 +1,19 @@ package mxgateway import ( + "errors" "fmt" pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated" ) +// ErrSlowConsumer is the terminal error sent on the Events/EventsAfter +// (cancel-when-full) path when the buffered results channel overflows because +// the consumer fell behind. It is delivered as the final EventResult.Err before +// the channel closes, so overflow is always observable rather than silently +// dropping events. Match it with errors.Is. +var ErrSlowConsumer = errors.New("mxgateway: event consumer fell behind; stream terminated") + // GatewayError wraps transport-level gRPC failures. type GatewayError struct { // Op names the operation that failed (for example "dial" or "invoke"). diff --git a/clients/go/mxgateway/session.go b/clients/go/mxgateway/session.go index 00bcdab..e61cf46 100644 --- a/clients/go/mxgateway/session.go +++ b/clients/go/mxgateway/session.go @@ -18,6 +18,15 @@ import ( const maxBulkItems = 1000 +// eventBufferSize is the number of buffered event slots on the +// Events/EventsAfter (cancel-when-full) results channel. +const eventBufferSize = 16 + +// eventBufferReservedSlots is the extra capacity reserved beyond +// eventBufferSize so a terminal ErrSlowConsumer result can always be enqueued +// non-blockingly on overflow, even when all data slots are full. +const eventBufferReservedSlots = 1 + // EventResult carries either the next ordered event or a terminal stream error. type EventResult struct { // Event is the next event from the stream when Err is nil. @@ -674,11 +683,22 @@ func (s *Session) Write2Raw(ctx context.Context, serverHandle, itemHandle int32, // Events streams ordered session events until the server ends the stream, // context cancellation stops Recv, or a terminal error is sent. +// +// The returned channel is buffered. If the consumer falls behind and the buffer +// overflows, the stream is terminated and a final EventResult carrying a +// GatewayError that wraps ErrSlowConsumer is delivered before the channel +// closes. Callers must match it with errors.Is(res.Err, ErrSlowConsumer) to +// distinguish a slow-consumer drop from a graceful server end. Use +// SubscribeEvents for a blocking, backpressured stream that never drops. func (s *Session) Events(ctx context.Context) (<-chan EventResult, error) { return s.EventsAfter(ctx, 0) } // EventsAfter streams ordered session events after the given worker sequence. +// +// Like Events, the returned channel is buffered and terminates with a final +// EventResult wrapping ErrSlowConsumer (matchable via errors.Is) if the consumer +// falls behind and the buffer overflows, rather than silently closing. func (s *Session) EventsAfter(ctx context.Context, afterWorkerSequence uint64) (<-chan EventResult, error) { subscription, err := s.subscribeEventsAfter(ctx, afterWorkerSequence, true) if err != nil { @@ -708,7 +728,7 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence return nil, err } - results := make(chan EventResult, 16) + results := make(chan EventResult, eventBufferSize+eventBufferReservedSlots) done := make(chan struct{}) go func() { defer close(results) @@ -756,14 +776,30 @@ func sendEventResult( cancel context.CancelFunc, ) bool { if cancelWhenBufferFull { + // Treat the channel as full once the eventBufferSize data slots are + // occupied, keeping eventBufferReservedSlots free for the terminal + // error. This goroutine is the sole producer, so len(results) only + // grows by our own sends and shrinks as the consumer reads; the reserve + // therefore always survives to carry ErrSlowConsumer. A plain buffered + // send would instead consume the reserved slot as ordinary data. + if len(results) >= eventBufferSize { + // The consumer fell behind and the data slots are full. Cancel the + // stream, then use the reserved terminal slot to enqueue a loud + // ErrSlowConsumer result so overflow is always observable rather + // than a silently closed channel indistinguishable from a graceful + // server end. + cancel() + select { + case results <- EventResult{Err: &GatewayError{Op: "stream events", Err: ErrSlowConsumer}}: + default: + } + return false + } select { case results <- result: return true case <-ctx.Done(): return false - default: - cancel() - return false } } From 5faef6c012158e59fd7b9ac15c4a43166b34fca7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:45 -0400 Subject: [PATCH 2/5] fix(CLI-03): Rust typed invoke validates HRESULT/MXSTATUS_PROXY Typed write/command wrappers now call ensure_mxaccess_success after ensure_command_success, failing on hresult < 0 (COM-correct, matching CLI-08/Python) or any MxStatusProxy.success == 0, via a new boxed Error::MxAccess variant with credential-safe formatting. The raw invoke escape hatch stays unvalidated. archreview: CLI-03 (P0). Verified: cargo fmt/check/test (28) and clippy --all-targets -D warnings clean. --- clients/rust/README.md | 12 +++ clients/rust/src/client.rs | 20 +++-- clients/rust/src/error.rs | 168 ++++++++++++++++++++++++++++++++++++- clients/rust/src/lib.rs | 2 +- 4 files changed, 194 insertions(+), 8 deletions(-) diff --git a/clients/rust/README.md b/clients/rust/README.md index e01e645..4548fee 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -125,6 +125,18 @@ preserving the raw message for parity diagnostics. Command replies whose protocol status is not `PROTOCOL_STATUS_CODE_OK` become `Error::Command` and retain the raw `MxCommandReply`. +The typed command helpers (`register`, `add_item`, `write`, the bulk variants, +etc.) also enforce MXAccess parity on an otherwise-OK reply: a reply that +reports a negative `hresult` (COM failure semantics — a positive code such as +`S_FALSE = 1` is a success) or a non-success `MXSTATUS_PROXY` status entry +becomes `Error::MxAccess`, which boxes an `MxAccessError` retaining the raw +`MxCommandReply` (recover it with `MxAccessError::reply` / `into_reply`). Its +message summarizes the `hresult` and status entries with credential-safe +redaction. Per-item bulk failures are reported inside each result entry +(`was_successful = false`) and do not raise `Error::MxAccess`. The raw +`invoke_raw` / `client.invoke_raw` escape hatch performs neither check and +returns the unvalidated reply. + ## Write Semantics And Common Pitfalls These are MXAccess parity behaviors that surprise new callers. The gateway diff --git a/clients/rust/src/client.rs b/clients/rust/src/client.rs index bdac543..493503c 100644 --- a/clients/rust/src/client.rs +++ b/clients/rust/src/client.rs @@ -11,7 +11,9 @@ use tonic::transport::Channel; use tonic::Request; use crate::auth::AuthInterceptor; -use crate::error::{ensure_command_success, ensure_protocol_success, Error}; +use crate::error::{ + ensure_command_success, ensure_mxaccess_success, ensure_protocol_success, Error, +}; use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGatewayClient; use crate::generated::mxaccess_gateway::v1::{ AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AlarmFeedMessage, @@ -166,16 +168,24 @@ impl GatewayClient { Ok(response.into_inner()) } - /// Issue an `Invoke` RPC and surface a non-OK reply as - /// [`Error::Command`]. + /// Issue an `Invoke` RPC and surface a failing reply as a typed error. + /// + /// The reply is validated twice: [`ensure_command_success`] rejects a + /// non-OK protocol envelope as [`Error::Command`], then + /// [`ensure_mxaccess_success`] rejects an MXAccess-level failure (negative + /// `hresult` or a non-success `MXSTATUS_PROXY` entry) as + /// [`Error::MxAccess`], preserving MXAccess parity. Callers that need the + /// unvalidated reply should use [`invoke_raw`](Self::invoke_raw) instead. /// /// # Errors /// /// Returns [`Error::Command`] when the reply's `protocol_status` is not - /// `Ok`, plus any errors propagated by + /// `Ok`, [`Error::MxAccess`] when the reply reports an MXAccess-level + /// failure, plus any errors propagated by /// [`invoke_raw`](Self::invoke_raw). pub async fn invoke(&self, request: MxCommandRequest) -> Result { - ensure_command_success(self.invoke_raw(request).await?) + let reply = ensure_command_success(self.invoke_raw(request).await?)?; + ensure_mxaccess_success(reply) } /// Open the server-streaming `StreamEvents` RPC. diff --git a/clients/rust/src/error.rs b/clients/rust/src/error.rs index e7c8124..0c335b9 100644 --- a/clients/rust/src/error.rs +++ b/clients/rust/src/error.rs @@ -9,7 +9,9 @@ use thiserror::Error as ThisError; use tonic::Code; -use crate::generated::mxaccess_gateway::v1::{MxCommandReply, ProtocolStatus, ProtocolStatusCode}; +use crate::generated::mxaccess_gateway::v1::{ + MxCommandReply, MxStatusCategory, ProtocolStatus, ProtocolStatusCode, +}; /// Top-level error type returned by the Rust client wrappers. /// @@ -95,6 +97,15 @@ pub enum Error { #[error("gateway command failed: {0}")] Command(#[from] Box), + /// Gateway accepted the call and returned an `Ok` protocol envelope, but + /// the reply reported an MXAccess-level failure — a negative `hresult` + /// (COM failure semantics) or one or more `MXSTATUS_PROXY` entries that + /// did not indicate success. The wrapped [`MxAccessError`] preserves the + /// full reply so callers can inspect the native status payload. Boxed to + /// keep the containing enum small, matching [`Error::Command`]. + #[error("gateway command reported an MXAccess failure: {0}")] + MxAccess(#[from] Box), + /// Protocol-level operation (open/close session) returned a non-OK /// [`ProtocolStatus`] envelope. #[error("gateway {operation} failed: {code:?}: {message}")] @@ -175,6 +186,72 @@ impl std::fmt::Display for CommandError { impl std::error::Error for CommandError {} +/// Wrapper around an [`MxCommandReply`] whose protocol envelope succeeded but +/// whose MXAccess-level result reported a failure — a negative `hresult` or a +/// non-success `MXSTATUS_PROXY` entry. +/// +/// The wrapper is heap-allocated inside [`Error::MxAccess`] to keep the +/// containing enum small. Callers can recover the reply with +/// [`MxAccessError::reply`] or [`MxAccessError::into_reply`]. Its `Display` +/// summarizes the `hresult` and status entries and scrubs any credential-like +/// tokens from diagnostic text before it reaches a caller. +#[derive(Clone, Debug)] +pub struct MxAccessError { + reply: MxCommandReply, +} + +impl MxAccessError { + /// Wrap a reply whose MXAccess-level result reported a failure. + pub fn new(reply: MxCommandReply) -> Self { + Self { reply } + } + + /// Borrow the underlying reply (correlation id, hresult, statuses). + pub fn reply(&self) -> &MxCommandReply { + &self.reply + } + + /// Consume the error and return the underlying reply. + pub fn into_reply(self) -> MxCommandReply { + self.reply + } +} + +impl std::fmt::Display for MxAccessError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let hresult = match self.reply.hresult { + Some(value) => value.to_string(), + None => "none".to_owned(), + }; + + write!( + formatter, + "hresult={hresult}, {} status entr{}", + self.reply.statuses.len(), + if self.reply.statuses.len() == 1 { + "y" + } else { + "ies" + } + )?; + + for status in &self.reply.statuses { + let category = MxStatusCategory::try_from(status.category) + .unwrap_or(MxStatusCategory::Unspecified); + let diagnostic = redact_credentials(&status.diagnostic_text); + write!( + formatter, + "; [success={}, category={category:?}, detail={}, {}]", + status.success, status.detail, diagnostic + )?; + } + + Ok(()) + } +} + +impl std::error::Error for MxAccessError {} + impl From for Error { fn from(status: tonic::Status) -> Self { let message = redact_credentials(status.message()); @@ -225,6 +302,36 @@ pub fn ensure_command_success(reply: MxCommandReply) -> Result Result { + let hresult_failure = reply.hresult.is_some_and(|hresult| hresult < 0); + let status_failure = reply.statuses.iter().any(|status| status.success == 0); + + if hresult_failure || status_failure { + Err(Box::new(MxAccessError::new(reply)).into()) + } else { + Ok(reply) + } +} + /// Validate a [`ProtocolStatus`] envelope returned by an open/close-session /// reply. /// @@ -271,7 +378,20 @@ fn redact_credentials(message: &str) -> String { mod tests { use tonic::{Code, Status}; - use super::Error; + use super::{ensure_mxaccess_success, Error}; + use crate::generated::mxaccess_gateway::v1::{ + MxCommandReply, MxStatusCategory, MxStatusProxy, ProtocolStatus, ProtocolStatusCode, + }; + + fn ok_reply() -> MxCommandReply { + MxCommandReply { + protocol_status: Some(ProtocolStatus { + code: ProtocolStatusCode::Ok as i32, + message: String::new(), + }), + ..MxCommandReply::default() + } + } #[test] fn classifies_authentication_status() { @@ -286,4 +406,48 @@ mod tests { assert!(message.contains("")); assert!(!message.contains("visible_secret")); } + + #[test] + fn ensure_mxaccess_success_passes_clean_reply() { + let mut reply = ok_reply(); + // Positive hresult (e.g. S_FALSE = 1) is a success, not a failure. + reply.hresult = Some(1); + reply.statuses = vec![MxStatusProxy { + success: 1, + category: MxStatusCategory::Ok as i32, + ..MxStatusProxy::default() + }]; + + assert!(ensure_mxaccess_success(reply).is_ok()); + } + + #[test] + fn ensure_mxaccess_success_flags_failing_status_entry() { + let mut reply = ok_reply(); + reply.statuses = vec![MxStatusProxy { + success: 0, + category: MxStatusCategory::CommunicationError as i32, + detail: 42, + diagnostic_text: "write rejected for mxgw_visible_secret".to_owned(), + ..MxStatusProxy::default() + }]; + + let error = ensure_mxaccess_success(reply).expect_err("failing status must error"); + let message = error.to_string(); + + assert!(matches!(error, Error::MxAccess(_))); + assert!(message.contains("")); + assert!(!message.contains("visible_secret")); + } + + #[test] + fn ensure_mxaccess_success_flags_negative_hresult() { + let mut reply = ok_reply(); + // 0x80004005 (E_FAIL) as a signed 32-bit value. + reply.hresult = Some(-2_147_467_259); + + let error = ensure_mxaccess_success(reply).expect_err("negative hresult must error"); + + assert!(matches!(error, Error::MxAccess(_))); + } } diff --git a/clients/rust/src/lib.rs b/clients/rust/src/lib.rs index 38ffac8..44acce7 100644 --- a/clients/rust/src/lib.rs +++ b/clients/rust/src/lib.rs @@ -26,7 +26,7 @@ pub use auth::{ApiKey, AuthInterceptor}; #[doc(inline)] pub use client::{AlarmFeedStream, EventStream, GatewayClient}; #[doc(inline)] -pub use error::{CommandError, Error}; +pub use error::{CommandError, Error, MxAccessError}; #[doc(inline)] pub use galaxy::{DeployEventStream, GalaxyClient}; #[doc(inline)] From 31eec4145658be63935d18e331aa41376f37ff0c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:45 -0400 Subject: [PATCH 3/5] fix(WRK-01): STA pump step refreshes activity to prevent false StaHung A long legitimate ReadBulk pumped Windows messages without refreshing LastStaActivityUtc, so the watchdog false-positived StaHung past HeartbeatStuckCeiling and then silently dropped every reply. PumpPendingMessages() now calls MarkActivity() after pumping; a genuinely stuck STA (no pumping) still accrues staleness and faults correctly. No MXAccess parity change. archreview: WRK-01 (P0). Verified on the Windows host (x86): worker builds clean, StaRuntimeTests + WorkerPipeSessionTests 33/33 pass. --- docs/MxAccessWorkerInstanceDesign.md | 30 ++++-- .../Ipc/WorkerPipeSessionTests.cs | 96 +++++++++++++++++++ .../Sta/StaRuntimeTests.cs | 28 ++++++ .../Sta/StaRuntime.cs | 22 +++-- 4 files changed, 161 insertions(+), 15 deletions(-) diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index d490f59..77b6fa7 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -656,11 +656,15 @@ the event queue implementation owns those counters. The STA watchdog currently emits a `WorkerFault` with `WorkerFaultCategory.StaHung` when `LastStaActivityUtc` is older than `WorkerPipeSessionOptions.HeartbeatGrace` **and no command is in flight**. -`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` only immediately -before and after each work item, so a synchronously long-running STA command -(for example a `ReadBulk` waiting `timeout_ms` for the first `OnDataChange`) -legitimately freezes `LastStaActivityUtc` for the duration of the wait while -the worker is healthy. The watchdog is therefore suppressed while the +`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` immediately before +and after each work item, so a synchronously long-running STA command that +neither completes work items nor pumps would freeze `LastStaActivityUtc` for +the duration of the wait while the worker is healthy. Commands that hold the +STA to wait for COM events (for example a `ReadBulk` waiting `timeout_ms` for +the first `OnDataChange`) avoid this: they pump via +`StaRuntime.PumpPendingMessages()`, which now refreshes `LastStaActivityUtc` +on every iteration (see the `HeartbeatStuckCeiling` discussion below). The +watchdog is additionally suppressed while the heartbeat snapshot's `CurrentCommandCorrelationId` is non-empty: the worker is busy executing a command, not hung, and the heartbeat already surfaces the in-flight correlation id so the gateway can apply its own per-command timeout @@ -684,10 +688,18 @@ session and only the gateway's per-command timeout would catch the hang — losing the worker-originated diagnostic (`StaHung` fault category, the stale-by interval) from the gateway audit trail. Once `LastStaActivityUtc` has been stale for longer than `HeartbeatStuckCeiling`, the watchdog fires -`StaHung` regardless of whether a command is in flight, on the assumption -that no legitimate STA command should run that long without periodically -refreshing activity. Deployments that legitimately run very long bulk -operations should raise the ceiling rather than disable it. +`StaHung` regardless of whether a command is in flight. This is now safe for +healthy long-running commands: `StaRuntime.PumpPendingMessages()` refreshes +`LastStaActivityUtc` (via `MarkActivity()`) every time it runs, and long-hold +STA commands invoke it on every wait iteration (`ReadBulk` routes its +per-tag wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`). +A command that keeps pumping therefore keeps its activity timestamp fresh and +never reaches the ceiling, while a genuinely stuck STA — one that has stopped +pumping — accrues staleness and faults correctly. The ceiling is thus the +backstop for a command that both holds the thread and stops pumping, not a +guillotine for slow-but-healthy work. Deployments that legitimately run very +long bulk operations should still be able to raise the ceiling rather than +disable it. ## Shutdown diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 0c13b3a..117770a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -704,6 +704,102 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// WRK-01 regression: a long in-flight STA command that keeps pumping + /// must NOT self-fault as StaHung, and its reply must still be + /// delivered. The real fix makes StaRuntime.PumpPendingMessages + /// refresh LastActivityUtc on every wait iteration, so a healthy + /// ReadBulk holding the STA far longer than + /// HeartbeatStuckCeiling (75 s in production) keeps its activity + /// timestamp fresh. This test compresses the clock — a 100 ms ceiling + /// with a command in flight across a window many multiples longer — and + /// models the pump refresh by continuously advancing the snapshot's + /// LastStaActivityUtc while the command blocks. Contrast + /// , + /// where a frozen timestamp beyond the ceiling correctly faults; here + /// the refreshed timestamp must keep the fault suppressed and let the + /// reply through the Ready-state gate. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_LongInFlightCommandThatKeepsPumping_DoesNotFaultAndDeliversReply() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(20)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + BlockDispatch = true, + }; + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(20), + HeartbeatGrace = TimeSpan.FromMilliseconds(50), + HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(100), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + // Kick off the long command; it blocks in DispatchAsync until released, + // so its correlation id stays in flight in the heartbeat snapshot. + await pipePair.GatewayWriter + .WriteAsync(CreateCommandEnvelope("long-bulk-read"), cancellation.Token); + Assert.True( + runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(5)), + "The long command must reach the runtime and begin dispatch."); + + // Model the pump refreshing STA activity on each wait iteration: keep + // the snapshot's LastStaActivityUtc current while the command is in + // flight. + using CancellationTokenSource pumpRefresh = new(); + Task refreshLoop = Task.Run( + async () => + { + while (!pumpRefresh.IsCancellationRequested) + { + runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot( + DateTimeOffset.UtcNow, + pendingCommandCount: 1, + outboundEventQueueDepth: 0, + lastEventSequence: 0, + currentCommandCorrelationId: "long-bulk-read")); + await Task.Delay(TimeSpan.FromMilliseconds(20)).ConfigureAwait(false); + } + }); + + // Inspect a bounded number of frames over a window many multiples of the + // 100 ms ceiling (at least 30 heartbeats at 20 ms ~ 600 ms). None may be + // a WorkerFault while activity is continuously refreshed. + const int framesToInspect = 30; + for (int index = 0; index < framesToInspect; index++) + { + WorkerEnvelope envelope = await pipePair.GatewayReader + .ReadAsync(cancellation.Token); + Assert.NotEqual( + WorkerEnvelope.BodyOneofCase.WorkerFault, + envelope.BodyCase); + } + + // Stop refreshing and release the command; its reply must be delivered + // because the session never faulted (state stayed Ready). + pumpRefresh.Cancel(); + await refreshLoop; + runtime.ReleaseDispatch(); + + WorkerEnvelope reply = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.CorrelationId == "long-bulk-read", + cancellation.Token); + Assert.Equal( + ProtocolStatusCode.Ok, + reply.WorkerCommandReply.Reply.ProtocolStatus.Code); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + /// /// RunAsync must throw a diagnostic exception if the /// runtime-session factory returns null, rather than deferring the diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs index 3afaa85..f143324 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs @@ -85,6 +85,34 @@ public sealed class StaRuntimeTests Assert.True(updated); } + /// + /// Verifies that refreshes + /// . A long synchronous STA + /// command (for example ReadBulk waiting timeout_ms for + /// the first OnDataChange) invokes the pump step on every wait + /// iteration while it legitimately holds the STA thread; refreshing + /// activity here keeps the watchdog from mistaking a busy STA for a hung + /// one (WRK-01). The runtime is deliberately left unstarted so the only + /// source of activity is the pump call under test, not the idle loop. + /// + [Fact] + public void PumpPendingMessages_RefreshesLastActivity() + { + RecordingComApartmentInitializer initializer = new(); + using StaRuntime runtime = CreateRuntime(initializer); + DateTimeOffset before = runtime.LastActivityUtc; + + bool refreshed = SpinWait.SpinUntil( + () => + { + runtime.PumpPendingMessages(); + return runtime.LastActivityUtc > before; + }, + TimeSpan.FromSeconds(2)); + + Assert.True(refreshed, "PumpPendingMessages must advance LastActivityUtc."); + } + /// Verifies that InvokeAsync faults the returned task when a command raises an exception without stopping the runtime. /// A task that represents the asynchronous operation. [Fact] diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs index 2ba2d37..f8a83c1 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs @@ -80,14 +80,24 @@ public sealed class StaRuntime : IDisposable public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet; /// - /// Pumps any pending Windows messages on the calling thread. Intended - /// for commands that synchronously hold the STA (e.g. ReadBulk) and - /// must allow inbound MXAccess COM events to dispatch while they - /// wait. Callers must already be on the STA; the method is otherwise - /// safe (PeekMessage simply finds no messages). + /// Pumps any pending Windows messages on the calling thread and refreshes + /// the STA activity timestamp. Intended for commands that synchronously + /// hold the STA (e.g. ReadBulk) and must allow inbound MXAccess COM events + /// to dispatch while they wait. Because a long-running command invokes this + /// on every wait iteration, refreshing activity here keeps a busy STA from + /// being mistaken for a hung one: a healthy command that keeps pumping stays + /// fresh past HeartbeatStuckCeiling, while a genuinely stuck STA (no + /// pumping) still accrues staleness and faults correctly. Callers must + /// already be on the STA; the method is otherwise safe (PeekMessage simply + /// finds no messages). /// /// The number of messages pumped. - public int PumpPendingMessages() => messagePump.PumpPendingMessages(); + public int PumpPendingMessages() + { + int pumpedMessages = messagePump.PumpPendingMessages(); + MarkActivity(); + return pumpedMessages; + } /// /// Starts the STA thread. From 20392cf246c71fee4bbeb7a226386488c7e9bf1a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:57 -0400 Subject: [PATCH 4/5] 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; From f7264580861d5ac6feeae3b066fdedd12b6ad0d8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:57 -0400 Subject: [PATCH 5/5] docs(archreview): mark P0 tier Done in remediation tracker All 8 P0 findings (GWC-01/02/03, WRK-01, CLI-01, CLI-03, TST-02, TST-12) recorded as Done with verification notes and change-log entries. --- archreview/remediation/00-tracking.md | 34 ++++++++++++++------------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/archreview/remediation/00-tracking.md b/archreview/remediation/00-tracking.md index 19a439e..a6e5cb8 100644 --- a/archreview/remediation/00-tracking.md +++ b/archreview/remediation/00-tracking.md @@ -42,14 +42,14 @@ Tiers follow the roadmap in `00-overall.md`. Work top-down; within a tier, `Dep` | ID | Sev | Eff | Dep | Status | Title | |---|---|:-:|---|---|---| -| GWC-01 | Critical | M | — | Not started | Alarm monitor and distributor pump both drain the single worker event channel | -| GWC-02 | High | M | — | Not started | Faulted sessions are never swept (worker + slot pinned up to 30 min) | -| GWC-03 | High | S | — | Not started | Documented sparse-array max-length bound is unimplemented | -| WRK-01 | High | M | — | Not started | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped | -| CLI-01 | High | M | — | Not started | Go `Session.Events()` silently closes stream on 16-slot overflow | -| CLI-03 | High | M | — | Not started | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY | -| TST-02 | High | M | TST-04 | Not started | Reconnect owner re-validation not implemented (security control) | -| TST-12 | Medium | S | — | Not started | CLAUDE.md misstates default retention behaviour (defaults are on, not off) | +| GWC-01 | Critical | M | — | Done | Alarm monitor and distributor pump both drain the single worker event channel | +| GWC-02 | High | M | — | Done | Faulted sessions are never swept (worker + slot pinned up to 30 min) | +| GWC-03 | High | S | — | Done | Documented sparse-array max-length bound is unimplemented | +| WRK-01 | High | M | — | Done | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped | +| CLI-01 | High | M | — | Done | Go `Session.Events()` silently closes stream on 16-slot overflow | +| CLI-03 | High | M | — | Done | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY | +| TST-02 | High | M | TST-04 | Done | Reconnect owner re-validation not implemented (security control) | +| TST-12 | Medium | S | — | Done | CLAUDE.md misstates default retention behaviour (defaults are on, not off) | ## Finding registers by domain @@ -59,9 +59,9 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| GWC-01 | Critical | P0 | M | — | Not started | Alarm monitor and distributor pump both drain the single worker event channel | -| GWC-02 | High | P0 | M | — | Not started | Faulted sessions are never swept | -| GWC-03 | High | P0 | S | — | Not started | Documented sparse-array max-length bound is unimplemented | +| GWC-01 | Critical | P0 | M | — | Done | Alarm monitor and distributor pump both drain the single worker event channel | +| GWC-02 | High | P0 | M | — | Done | Faulted sessions are never swept | +| GWC-03 | High | P0 | S | — | Done | Documented sparse-array max-length bound is unimplemented | | GWC-04 | Medium | P1 | M | — | Not started | Full event channel stalls the worker read loop behind command replies | | GWC-05 | Medium | — | S | — | Not started | Worker pipe created with no ACL / no CurrentUserOnly | | GWC-06 | Medium | P2 | S | GWC-07,08 | Not started | Stopwatch allocated per streamed event | @@ -87,7 +87,7 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| WRK-01 | High | P0 | M | — | Not started | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped | +| WRK-01 | High | P0 | M | — | Done | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped | | WRK-02 | Medium | — | M | — | Not started | STA thread death after startup is silent; future work hangs forever | | WRK-03 | Medium | — | S | WRK-02 | Not started | Commands after shutdown starts are dropped with no reply | | WRK-04 | Medium | — | M | — | Not started | Envelope `sequence` can appear out of order on the wire | @@ -174,9 +174,9 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| CLI-01 | High | P0 | M | — | Not started | Go `Session.Events()` silently closes stream on 16-slot overflow | +| CLI-01 | High | P0 | M | — | Done | Go `Session.Events()` silently closes stream on 16-slot overflow | | CLI-02 | High | P1 | M | — | Not started | Rust crate unbuildable outside the repo (`build.rs` path + `--no-verify`) | -| CLI-03 | High | P0 | M | — | Not started | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY | +| CLI-03 | High | P0 | M | — | Done | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY | | CLI-04 | High | P2 | L | CLI-15 | Not started | Typed-command parity gap across all five clients | | CLI-05 | Medium | — | S | — | Not started | .NET session cannot be re-attached to an existing session id | | CLI-06 | Medium | — | S | — | Not started | .NET `DisposeAsync` blocks/throws on unreachable gateway | @@ -214,7 +214,7 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| | TST-01 | High | P2 | L | TST-04 | Not started | Reconnect/replay has no e2e test and no client consumer | -| TST-02 | High | P0 | M | TST-04 | Not started | Reconnect owner re-validation not implemented | +| TST-02 | High | P0 | M | TST-04 | Done | Reconnect owner re-validation not implemented | | TST-03 | High | P1 | M | — | Not started | No CI exists | | TST-04 | High | P2 | L | — | Not started | Session-resilience epic 16/28 tasks unfinished | | TST-05 | Medium | P1 | S | TST-03 | Not started | Real-worker control/COM paths verified opt-in only | @@ -224,7 +224,7 @@ Full design + implementation for each row lives in the linked domain doc under i | TST-09 | Medium | — | M | — | Not started | Health checks cover only the auth store | | TST-10 | Medium | — | M | — | Not started | Deployment/upgrade undocumented and hand-rolled | | TST-11 | Medium | P2 | M | — | Not started | No version discipline on server; client versions drift | -| TST-12 | Medium | P0 | S | — | Not started | CLAUDE.md misstates default retention behaviour | +| TST-12 | Medium | P0 | S | — | Done | CLAUDE.md misstates default retention behaviour | | TST-13 | Medium | P2 | S | — | Not started | gateway.md carries stale design-era sketches | | TST-14 | Medium | P2 | S | — | Not started | Repo-root working artifacts need triage | | TST-15 | Medium | P2 | M | TST-04 | Not started | Dashboard EventsHub has no per-session ACL | @@ -254,3 +254,5 @@ Findings the review flagged as one coordinated design pass — sequence them tog | Date | Change | |---|---| | 2026-07-09 | Initial tracking doc generated from the six domain remediation designs. All 153 findings `Not started`. | +| 2026-07-09 | P0 tier executed via parallel agents. GWC-01/02/03, TST-02, TST-12, CLI-01, CLI-03 → `Done` (NonWindows build clean; gateway 135/135 targeted tests pass, Go `go test ./...` + `-race` pass, Rust `test`/`clippy -D warnings` pass). | +| 2026-07-09 | WRK-01 → `Done`. Verified on Windows host (windev) via an isolated `origin/main` worktree: worker builds x86 clean, `StaRuntimeTests`+`WorkerPipeSessionTests` 33/33 pass. Fixed an `xUnit1030` build error (the new worker test used `.ConfigureAwait(false)` in `[Fact]` bodies) that the macOS tree could not surface. Also ran GWC-01's Windows-only `WorkerClientTests` on windev: 18/18 pass (incl. `ReadEventsAsync_SecondEnumerator_Throws`). All 8 P0 findings now `Done`. Not yet committed. |