diff --git a/docs/AlarmClientDiscovery.md b/docs/AlarmClientDiscovery.md index ad40a4b..5ec7aa8 100644 --- a/docs/AlarmClientDiscovery.md +++ b/docs/AlarmClientDiscovery.md @@ -1051,6 +1051,18 @@ Every transition and snapshot record carries `degraded = true` and `source_provider = ALARM_PROVIDER_MODE_SUBTAG`. Clients that require full fidelity must wait for failback to the alarm manager. +### Retrigger transitions + +`ALARM_TRANSITION_KIND_RETRIGGER` is reserved in the contract and the worker +never emits it, in either provider mode. The alarmmgr XML reports only the +four states `UNACK_ALM`, `ACK_ALM`, `UNACK_RTN`, and `ACK_RTN`, so a condition +that returns to alarm after clearing (`*_RTN` → `*_ALM`) looks the same as a +new raise, and `AlarmRecordTransitionMapper.MapTransition` reports it as +`ALARM_TRANSITION_KIND_RAISE`. The subtag source has no re-alarm counter to +observe either. The worker does not infer a retrigger from timing or +from a raise that follows a clear, because the gateway forwards only +transitions its source reports. + ### Provider mode reflection When `FailoverAlarmConsumer` switches between providers, it raises diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 2c0f4b3..b829364 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -133,6 +133,9 @@ starts the STA, or touches MXAccess: | `2` | `InvalidArguments` | Required arguments are missing or unknown arguments are present. | | `3` | `InvalidProtocolVersion` | `--protocol-version` is not numeric or does not match the supported worker protocol. | | `4` | `MissingNonce` | `MXGATEWAY_WORKER_NONCE` is absent or empty. | +| `5` | `PipeConnectionFailed` | An `IOException` or `TimeoutException` escapes the pipe client. | +| `6` | `ProtocolViolation` | A `WorkerFrameProtocolException` escapes the pipe client. | +| `7` | `UnhandledException` | An exception reaches the `AppDomain.UnhandledException` hook (see `docs/WorkerBootstrap.md`). | Bootstrap logs use `WorkerConsoleLogger` key/value output. `WorkerLogRedactor` redacts fields whose names indicate nonce, secret, password, token, @@ -270,12 +273,33 @@ is still responsive. Shutdown marks the runtime as closing, wakes the pump, rejects new commands, cancels queued work, uninitializes COM on the STA, and waits for the thread to exit. +`ProcessQueuedCommands` pumps pending Windows messages between queued work +items whenever more work remains, not only after the whole queue drains. A +burst of queued commands would otherwise hold every inbound COM callback +(`OnDataChange`, `OnWriteComplete`) behind the entire backlog. + +If the STA loop itself throws after `Start` has returned (a failed +`MsgWaitForMultipleObjectsEx`, or an exception escaping a dispatched message), +`StaRuntime` records the exception as `LoopFault`, fails every later +`InvokeAsync` with an `InvalidOperationException` wrapping it (never the +graceful `StaRuntimeShutdownException`), cancels queued work, and invokes the +handler registered through `RegisterLoopFaultHandler` — unless a requested +shutdown is already in progress, in which case the death is part of a +deliberate teardown and is not reported. `MxAccessStaSession` registers a +handler that records a `StaHung` `WorkerFault` on the event queue (diagnostic +text carries the exception type and HRESULT only, never the message), +so the IPC drain loop sends the fault to the gateway and ends the pipe session +and the worker process. Before this, a post-start loop death was stored in the +startup-exception field that only `Start` reads, leaving a live worker whose +commands queued forever. Startup failures still surface from `Start` itself. + ### Inner Completion Waits Two commands hold the STA while waiting for a COM event they just provoked: the unary write path waits for its `OnWriteComplete` -(`MxAccessWriteCompletionCache.TryWaitForCompletion`), and `ReadBulk` waits per -tag for the first `OnDataChange` (`MxAccessValueCache.TryWaitForUpdate`). Both +(`MxAccessWriteCompletionCache.TryWaitForCompletion`), and `ReadBulk` waits for +the first `OnDataChange` of every snapshot tag +(`MxAccessValueCache.TryWaitForUpdates`). Both run the same loop shape as the outer pump, for the same reason — the event they are waiting for *is* a Windows message, so the thread must keep dispatching to receive it: @@ -297,7 +321,7 @@ The idle slice is a Win32 wait (`StaWaitHelper.WaitForSignalOrMessages`), never `Thread.Sleep`. A sleeping STA pumps no messages, so a sleep-polled loop could only dispatch the awaited COM event at poll-tick granularity while stalling *every other* event for the same tick — up to 1.5 s for a write completion and -up to `timeout_ms` per tag for `ReadBulk`. The Win32 wait returns the instant a +up to `timeout_ms` for `ReadBulk`. The Win32 wait returns the instant a message needs pumping, so the apartment dispatches continuously for the whole wait. Each cache also sets an `AutoResetEvent` from its update path (outside the cache lock) so a cross-thread producer wakes the waiter immediately; in the live @@ -317,8 +341,29 @@ nothing wakes the wait — a process with no STA message queue (unit tests drive these caches from ordinary threads, standing in for the STA by updating the cache from a fake `pumpStep`) must not block for a full poll interval. Timeouts, deadline math, and return values are unchanged by the wait mechanism: an expired -write wait still yields the empty-`statuses` unconfirmed reply, and an expired -per-tag `ReadBulk` wait still reports its own timeout. +write wait still yields the empty-`statuses` unconfirmed reply, and every +`ReadBulk` tag that did not deliver before the deadline still reports its own +timeout. + +`ReadBulk` runs its snapshot lifecycle in phases so the STA is held for at most +one `timeout_ms` per command, not one per tag: (1) every tag is validated and +probed against the cached-and-advised fast path; (2) every remaining tag is +added and advised; (3) one wait with a shared deadline collects the first +`OnDataChange` per snapshot handle and returns as soon as all have delivered; +(4) every snapshot handle is unadvised and removed, best-effort. Per-tag +results are unchanged: request order, the blank-address failure, the cached +result, an `AddItem` failure (item handle 0) or `Advise` failure (its own item +handle) failing only that tag, and the `ReadBulk timed out after ms waiting +for first OnDataChange.` message for each tag that never delivered. The cache +probes all run before any snapshot `AddItem`, so a later tag in the same +request can never be served from an earlier tag's snapshot subscription. +When MXAccess returns the same item handle for two tags in one batch (a +duplicate address), the handle is advised, unadvised, and removed once, and +every duplicate receives the shared outcome. Phases 2 and 4 make one or two +COM calls per tag without waiting, so they call the pump step every +`ReadBulkPumpInterval` (16) tags; a large batch therefore keeps dispatching +inbound callbacks and refreshing `LastStaActivityUtc` instead of drifting +toward `HeartbeatStuckCeiling`. The write wait's budget is `MxGateway:Worker:WriteCompletionWaitMilliseconds` (default 1500). It applies to `Write`, `Write2`, `WriteSecured`, and @@ -671,7 +716,20 @@ The worker records plain and supervisory advice separately only after the COM call returns normally. Successful `UnAdvise` removes all tracked advice for the server and item pair because the public MXAccess cleanup method has no plain versus supervisory selector. Successful `RemoveItem` and `Unregister` also clear -related advice state from the worker registry. Failed advice and cleanup calls +related advice state from the worker registry, and evict the per-handle +caches the event sink fills: `RemoveItem` drops the handle pair's +`MxAccessValueCache` and `MxAccessWriteCompletionCache` entries, and +`Unregister` drops every entry recorded under the server handle (MXAccess +removes the server's items with it, so the per-item eviction never runs for +them). Without this, an `AddItem` → `Write` → `RemoveItem` churn grew the +completion cache by one entry per cycle for the life of the worker. +`AddItem`, `AddItem2`, and `AddBufferedItem` also drop both cache entries for +the handle they receive when the registry does not already track it. An +`OnDataChange` or `OnWriteComplete` that was queued when `RemoveItem` or +`Unregister` ran is dispatched after the eviction and recreates the entry; +when MXAccess later reuses that item id, `ReadBulk`'s cached path would +otherwise serve the stale value as `was_cached = true`. A handle the +registry still tracks belongs to a live item, so its entries are kept. Failed advice and cleanup calls leave registry state unchanged so diagnostics continue to reflect the last successful MXAccess-owned state transition. @@ -905,6 +963,23 @@ the STA more than 75 s without pumping *should* fault — that is the ceiling's contract). The flag is named generically so any future non-dispatcher STA work reuses it. +A `PollOnce` that throws does not fault the session on the first failure. The +poll loop counts consecutive failures and records the alarm-poll fault +(`MxaccessEventConversionFailed`) only when the count reaches +`MxAccessStaSession.DefaultAlarmPollFailureThreshold` (3, the same default as +`AlarmFailoverConfig.consecutive_failure_threshold`); a successful poll resets +the count. Each failure below the threshold writes one +`level=Warning event=AlarmPollFailed` line to stderr carrying the exception +type, HRESULT, and counts only — the exception message is left out because it +can echo alarm or tag names. A single transient COM failure in +`GetXmlCurrentAlarms2` used to fault the whole session, taking every data +subscription with it; a persistent failure, including an STA-affinity +regression that fails every tick, still faults within three poll intervals. +A poll failure seen after the STA loop itself has died (`StaRuntime.LoopFault` +is set) ends the poll loop without recording anything: the loop-fault handler +has already recorded the `StaHung` fault, and the failure is only the +fast-fail `InvokeAsync` returns afterwards. + The in-flight-command suppression itself is bounded by `WorkerPipeSessionOptions.HeartbeatStuckCeiling` (default 75 seconds = 5 × `HeartbeatGrace`). The motivating case for the suppression is a legitimately diff --git a/docs/WorkerBootstrap.md b/docs/WorkerBootstrap.md index aa2ffc6..73aa9bd 100644 --- a/docs/WorkerBootstrap.md +++ b/docs/WorkerBootstrap.md @@ -117,8 +117,13 @@ public static WorkerBootstrapResult Failure(WorkerExitCode exitCode, IEnumerable | `MissingNonce` | 4 | The `MXGATEWAY_WORKER_NONCE` environment variable is null, empty, or whitespace. | | `PipeConnectionFailed` | 5 | An `IOException` or `TimeoutException` escapes the pipe client. | | `ProtocolViolation` | 6 | A `WorkerFrameProtocolException` escapes the pipe client. | +| `UnhandledException` | 7 | An exception reaches `AppDomain.CurrentDomain.UnhandledException` — thrown on a thread outside `WorkerApplication.Run`'s `try/catch` (a background task continuation, a timer, a COM callback thread). | -`InvalidArguments`, `InvalidProtocolVersion`, and `MissingNonce` originate in the parser; the others originate in `WorkerApplication.Run`'s `try/catch` around the pipe client. +`InvalidArguments`, `InvalidProtocolVersion`, and `MissingNonce` originate in the parser; `UnhandledException` originates in the process-wide hook described below; the others originate in `WorkerApplication.Run`'s `try/catch` around the pipe client. + +### Unhandled-exception hook + +The production entry point, `WorkerApplication.Run(string[])`, subscribes `AppDomain.CurrentDomain.UnhandledException` before it parses arguments. The handler (`WorkerApplication.HandleUnhandledException`) logs one `WorkerUnhandledException` error line and calls `Environment.Exit(7)`. Without it, an exception on a thread that `Run` does not own ended the process with the CLR's crash code and no worker log line, so the gateway saw an unexplained exit. The line carries `exit_code`, `exception_type`, and `is_terminating` only: the exception message and stack are left out because they can carry tag values or credentials passed through a command, the same rule every other worker failure line follows. A failure inside the logger is swallowed so the exit always happens. The injectable `Run` overloads that tests call do not install the hook, because it is process-global. ## Environment Abstraction diff --git a/docs/WorkerSta.md b/docs/WorkerSta.md index 8d20685..84de4eb 100644 --- a/docs/WorkerSta.md +++ b/docs/WorkerSta.md @@ -86,10 +86,16 @@ StaWorkItem workItem = new(command, cancellationToken); lock (gate) { + if (loopFault is not null) + { + return Task.FromException(new InvalidOperationException( + "The worker STA loop terminated unexpectedly.", + loopFault)); + } + if (shutdownRequested) { - return Task.FromException( - new InvalidOperationException("The worker STA runtime is shutting down.")); + return Task.FromException(new StaRuntimeShutdownException()); } commandQueue.Enqueue(workItem); @@ -101,7 +107,11 @@ return workItem.Task; `StaWorkItem` uses an `Interlocked.CompareExchange` on `started` so that exactly one of three outcomes happens: the STA thread executes the delegate, an external `CancellationToken` cancellation fires first, or `CancelBeforeExecution` runs during shutdown. `Execute` runs on the STA thread, sets `Completion` from `command()`, and propagates exceptions through `TrySetException` so the awaiting caller observes them. -`ProcessQueuedCommands` is the only consumer of the queue and runs on the STA thread, so each work item is guaranteed to execute on the apartment that owns the COM object. +`ProcessQueuedCommands` is the only consumer of the queue and runs on the STA thread, so each work item is guaranteed to execute on the apartment that owns the COM object. It calls `PumpPendingMessages` between work items whenever more work is queued, so a backlog of commands cannot hold inbound COM callbacks until the whole queue drains. + +### Loop failure after start + +An exception thrown by the loop after `Start()` returned — `WaitForWorkOrMessages` reporting `WAIT_FAILED`, or an exception escaping a dispatched message — is not a startup failure, and nothing reads `startupException` after `Start()`. `ThreadMain` instead stores it as `LoopFault` under `gate`, so every later `InvokeAsync` fails immediately with an `InvalidOperationException` whose inner exception is the loop fault, and work already queued is cancelled by the `finally` block. It then invokes the handler registered with `RegisterLoopFaultHandler`; a handler registered after the loop died runs immediately on the registering thread. Neither happens when `Shutdown` was already requested, because a loop that throws during a deliberate teardown is not a worker fault. `MxAccessStaSession` registers a handler that records a `StaHung` `WorkerFault` on the event queue, which the IPC drain loop sends to the gateway before ending the pipe session and the worker process. ## Command Dispatch diff --git a/gateway.md b/gateway.md index 9a6a8ec..7b61285 100644 --- a/gateway.md +++ b/gateway.md @@ -571,8 +571,15 @@ returns one result per requested entry — per-entry failures populate advised (`was_cached = true`, no subscription side-effects), or by taking a full `AddItem` + `Advise` + wait-for-first-OnDataChange + `UnAdvise` + `RemoveItem` snapshot lifecycle when no live subscription - exists (`was_cached = false`). Per-tag timeouts surface as - `was_successful = false` rather than throwing. The cache lives on the + exists (`was_cached = false`). `timeout_ms` is one shared deadline for + the whole command, not a per-tag budget: the worker adds and advises + every uncached tag first, waits once for all of them, then tears them + all down, so the STA is held for at most one `timeout_ms` (plus the + add/advise and teardown COM calls, during which it pumps every 16 tags). + Each tag that has not delivered by the deadline surfaces its own + `was_successful = false` timeout rather than throwing. Duplicate + addresses that MXAccess resolves to the same item handle share one + subscription and receive the same result. The cache lives on the worker's `MxAccessValueCache`, populated by `MxAccessBaseEventSink` on every `OnDataChange` after the event clears the outbound queue. Required scope: `invoke:read`. `timeout_ms == 0` uses the worker's diff --git a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs index d411a84..04bbdaa 100644 --- a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs @@ -1052,6 +1052,231 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output) } } + /// + /// TST-05: the real worker answers the four non-terminating control commands off-STA in + /// WorkerPipeSession.HandleControlCommandAsync. The default suite only ever sees the + /// FakeWorkerHarness canned replies, so this drives Ping, GetSessionState, + /// GetWorkerInfo, and DrainEvents through the gateway against a live worker and + /// asserts each reply carries real worker state rather than an INVALID_REQUEST. + /// + /// A task that represents the asynchronous operation. + [LiveMxAccessFact] + public async Task GatewaySession_WithLiveWorker_ControlCommands_ReturnRealWorkerState() + { + string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath(); + Assert.True( + File.Exists(workerExecutablePath), + $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}."); + + TestWorkerProcessFactory processFactory = new(output); + await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output); + + string? sessionId = null; + + try + { + OpenSessionReply openReply = await fixture.Service.OpenSession( + new OpenSessionRequest + { + ClientSessionName = "live-mxaccess-control", + ClientCorrelationId = "live-open-control", + CommandTimeout = Duration.FromTimeSpan(CommandTimeout), + }, + new TestServerCallContext()).ConfigureAwait(false); + + sessionId = openReply.SessionId; + Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code); + Assert.True(openReply.WorkerProcessId > 0); + + MxCommandReply pingReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-ping", new MxCommand + { + Kind = MxCommandKind.Ping, + Ping = new PingCommand { Message = "live-ping-echo" }, + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("Ping", pingReply); + Assert.Equal(ProtocolStatusCode.Ok, pingReply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.Ping, pingReply.Kind); + + MxCommandReply stateReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-get-session-state", new MxCommand + { + Kind = MxCommandKind.GetSessionState, + GetSessionState = new GetSessionStateCommand(), + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("GetSessionState", stateReply); + Assert.Equal(ProtocolStatusCode.Ok, stateReply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.GetSessionState, stateReply.Kind); + Assert.NotNull(stateReply.SessionState); + Assert.Equal(SessionState.Ready, stateReply.SessionState.State); + + MxCommandReply infoReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-get-worker-info", new MxCommand + { + Kind = MxCommandKind.GetWorkerInfo, + GetWorkerInfo = new GetWorkerInfoCommand(), + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("GetWorkerInfo", infoReply); + Assert.Equal(ProtocolStatusCode.Ok, infoReply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.GetWorkerInfo, infoReply.Kind); + Assert.NotNull(infoReply.WorkerInfo); + output.WriteLine( + $"GetWorkerInfo pid={infoReply.WorkerInfo.WorkerProcessId} version={infoReply.WorkerInfo.WorkerVersion} " + + $"progid={infoReply.WorkerInfo.MxaccessProgid} clsid={infoReply.WorkerInfo.MxaccessClsid}"); + Assert.Equal(openReply.WorkerProcessId, infoReply.WorkerInfo.WorkerProcessId); + Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.WorkerVersion)); + Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.MxaccessProgid)); + Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.MxaccessClsid)); + + // No StreamEvents subscriber is attached, so whatever the worker queued since start + // is still drainable; the reply must carry the DrainEvents payload (possibly empty). + MxCommandReply drainReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-drain-events", new MxCommand + { + Kind = MxCommandKind.DrainEvents, + DrainEvents = new DrainEventsCommand { MaxEvents = 16 }, + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("DrainEvents", drainReply); + Assert.Equal(ProtocolStatusCode.Ok, drainReply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.DrainEvents, drainReply.Kind); + Assert.Equal(MxCommandReply.PayloadOneofCase.DrainEvents, drainReply.PayloadCase); + Assert.True(drainReply.DrainEvents.Events.Count <= 16); + output.WriteLine($"DrainEvents drained={drainReply.DrainEvents.Events.Count}"); + + // A control command must not disturb the session: it is still Ready afterwards. + Assert.True(fixture.TryGetSession(sessionId, out GatewaySession? session)); + Assert.Equal(SessionState.Ready, session!.State); + } + finally + { + await ShutDownAsync(fixture, processFactory, sessionId, streamTask: null).ConfigureAwait(false); + } + } + + /// + /// TST-05: ShutdownWorker writes its OK reply before the worker shuts down, so the + /// reply reaches the caller (a reply written after the pipe closed would surface as a + /// pipe-disconnected failure instead), the worker process then exits on its own, and a later + /// command on the session does not succeed. Uses its own fixture and worker because it + /// terminates the worker. + /// + /// A task that represents the asynchronous operation. + [LiveMxAccessFact] + public async Task GatewaySession_WithLiveWorker_ShutdownWorker_RepliesOkThenWorkerExits() + { + string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath(); + Assert.True( + File.Exists(workerExecutablePath), + $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}."); + + TestWorkerProcessFactory processFactory = new(output); + await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output); + using RecordingServerStreamWriter eventWriter = new(); + + string? sessionId = null; + Task? streamTask = null; + using CancellationTokenSource streamCancellation = new(); + + try + { + OpenSessionReply openReply = await fixture.Service.OpenSession( + new OpenSessionRequest + { + ClientSessionName = "live-mxaccess-shutdown-worker", + ClientCorrelationId = "live-open-shutdown-worker", + CommandTimeout = Duration.FromTimeSpan(CommandTimeout), + }, + new TestServerCallContext()).ConfigureAwait(false); + + sessionId = openReply.SessionId; + Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code); + + // The gateway observes a worker's pipe closing through the event-stream read loop + // (the same path the abnormal-exit fact relies on); without a subscriber the session + // would sit in Ready until the heartbeat grace expired. + streamTask = fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId }, + eventWriter, + new TestServerCallContext(streamCancellation.Token)); + + MxCommandReply shutdownReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-shutdown-worker", new MxCommand + { + Kind = MxCommandKind.ShutdownWorker, + ShutdownWorker = new ShutdownWorkerCommand + { + GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(5)), + }, + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("ShutdownWorker", shutdownReply); + Assert.Equal(ProtocolStatusCode.Ok, shutdownReply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.ShutdownWorker, shutdownReply.Kind); + + // The worker exits on its own — no CloseSession, no kill. WaitForProcessesAsync + // throws on timeout, which fails the test. + await processFactory.WaitForProcessesAsync(StreamShutdownTimeout).ConfigureAwait(false); + + // The session can no longer serve commands once its worker is gone: either the + // gateway rejects the call outright or the reply is not Ok. + MxCommandReply? afterReply = null; + Exception? afterException = null; + try + { + afterReply = await fixture.Service.Invoke( + CreateControlRequest(sessionId, "live-ping-after-shutdown", new MxCommand + { + Kind = MxCommandKind.Ping, + Ping = new PingCommand(), + }), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("PingAfterShutdown", afterReply); + } + catch (Exception exception) + { + afterException = exception; + output.WriteLine($"PingAfterShutdown rejected: {exception.GetType().Name}: {exception.Message}"); + } + + Assert.True( + afterException is not null || afterReply!.ProtocolStatus.Code != ProtocolStatusCode.Ok, + "A command sent after ShutdownWorker must not succeed against the exited worker."); + + // Logged, not asserted: the gateway's WorkerClient moves to Closed on the shutdown + // ack, but GatewaySession.State is not driven from that transition today, so the + // public session state still reads Ready here (observed live on windev). That is a + // gateway-side gap, not a worker one — this fact pins the worker half of the + // contract: OK reply first, then a clean exit. + string observedState = fixture.TryGetSession(sessionId, out GatewaySession? session) + ? session.State.ToString() + : ""; + output.WriteLine($"ShutdownWorker observed_session_state={observedState}"); + } + finally + { + // The worker is gone; a CloseSession round-trip would only log a cleanup failure. + streamCancellation.Cancel(); + await ShutDownAsync(fixture, processFactory, sessionId: null, streamTask).ConfigureAwait(false); + } + } + + private static MxCommandRequest CreateControlRequest( + string sessionId, + string correlationId, + MxCommand command) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = correlationId, + Command = command, + }; + } + /// /// Closes the session and drains the event stream / worker processes without letting a /// cleanup timeout mask the original failure from the test body. diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Bootstrap/WorkerApplicationTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Bootstrap/WorkerApplicationTests.cs index 8742079..f29885e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Bootstrap/WorkerApplicationTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Bootstrap/WorkerApplicationTests.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using ZB.MOM.WW.MxGateway.Contracts; @@ -121,6 +122,49 @@ public sealed class WorkerApplicationTests Assert.Equal(typeof(InvalidOperationException).FullName, entry.Fields["exception_type"]); } + /// + /// WRK-09: the AppDomain unhandled-exception hook logs the exception type + /// (never its message) and exits with the dedicated exit code. + /// + [Fact] + public void HandleUnhandledException_LogsTypeOnlyAndExitsWithUnhandledExceptionCode() + { + MemoryWorkerLogger logger = new(); + int? exitCode = null; + + ZB.MOM.WW.MxGateway.Worker.WorkerApplication.HandleUnhandledException( + logger, + new InvalidOperationException("password=hunter2"), + isTerminating: true, + code => exitCode = code); + + Assert.Equal((int)WorkerExitCode.UnhandledException, exitCode); + Assert.Equal(7, (int)WorkerExitCode.UnhandledException); + MemoryWorkerLogEntry entry = Assert.Single(logger.Entries); + Assert.Equal("Error", entry.Level); + Assert.Equal("WorkerUnhandledException", entry.EventName); + Assert.Equal(typeof(InvalidOperationException).FullName, entry.Fields["exception_type"]); + Assert.True(entry.Fields["is_terminating"] is true); + Assert.DoesNotContain( + entry.Fields.Values, + value => value?.ToString()?.Contains("hunter2") == true); + } + + /// A failing logger must not prevent the defined exit. + [Fact] + public void HandleUnhandledException_WhenLoggerThrows_StillExits() + { + int? exitCode = null; + + ZB.MOM.WW.MxGateway.Worker.WorkerApplication.HandleUnhandledException( + new ThrowingLogger(), + exceptionObject: null, + isTerminating: false, + code => exitCode = code); + + Assert.Equal((int)WorkerExitCode.UnhandledException, exitCode); + } + private static string[] ValidArgs(string? protocolVersion = null) { return @@ -171,4 +215,15 @@ public sealed class WorkerApplicationTests throw _exception; } } + + private sealed class ThrowingLogger : IWorkerLogger + { + /// + public void Information(string eventName, IReadOnlyDictionary fields) => + throw new InvalidOperationException("logger failed"); + + /// + public void Error(string eventName, IReadOnlyDictionary fields) => + throw new InvalidOperationException("logger failed"); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmCommandHandlerTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmCommandHandlerTests.cs index 92c2d2c..597179b 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmCommandHandlerTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmCommandHandlerTests.cs @@ -499,6 +499,40 @@ public sealed class AlarmCommandHandlerTests Assert.Equal(AlarmProviderMode.Subtag, evt.OnAlarmProviderModeChanged.Mode); } + /// + /// In auto mode the primary consumer is built before the standby. When + /// building the standby throws, the primary is owned by nothing — the + /// failed Subscribe never reaches the dispatcher that would dispose it — + /// so the handler must dispose it rather than leak its COM object, and + /// the original failure must still propagate. + /// + [Fact] + public void Subscribe_AutoModeWhenStandbyBuildThrows_DisposesPrimaryAndRethrows() + { + FakeConsumer primary = new FakeConsumer(); + InvalidOperationException standbyFailure = new InvalidOperationException("standby build failed"); + AlarmCommandHandler handler = new AlarmCommandHandler( + new MxAccessEventQueue(), + () => primary, + threadAffinityCheck: null, + comFactory: null, + standbyFactory: _ => throw standbyFailure); + + SubscribeAlarmsCommand command = new SubscribeAlarmsCommand + { + SubscriptionExpression = @"\\HOST\Galaxy!Area", + ForcedMode = AlarmProviderMode.Unspecified, + }; + command.WatchList.Add(NewTarget("Galaxy!Area.Tank01.Level.HiHi")); + + InvalidOperationException thrown = Assert.Throws( + () => handler.Subscribe(command, "s1")); + + Assert.Same(standbyFailure, thrown); + Assert.True(primary.Disposed, "The primary consumer built before the failing standby must be disposed."); + Assert.False(handler.IsSubscribed); + } + /// An explicitly fallback-capable auto subscription builds a replaceable standby even when discovery is initially empty. [Fact] public void Subscribe_AutoModeWithEmptyWatchListAndFallbackEnabled_BuildsReplaceableStandby() diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmRecordTransitionMapperTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmRecordTransitionMapperTests.cs index 5009f6e..80c2f57 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmRecordTransitionMapperTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmRecordTransitionMapperTests.cs @@ -102,6 +102,22 @@ public sealed class AlarmRecordTransitionMapperTests Assert.Equal(expected, AlarmRecordTransitionMapper.MapTransition(previous, current)); } + /// + /// RETRIGGER is reserved and never synthesized: no state pair — including + /// cleared-to-active, the re-alarm shape — maps to it. + /// + [Fact] + public void MapTransition_ForEveryStatePair_NeverProducesRetrigger() + { + foreach (MxAlarmStateKind previous in System.Enum.GetValues(typeof(MxAlarmStateKind))) + { + foreach (MxAlarmStateKind current in System.Enum.GetValues(typeof(MxAlarmStateKind))) + { + Assert.NotEqual(AlarmTransitionKind.Retrigger, AlarmRecordTransitionMapper.MapTransition(previous, current)); + } + } + } + /// Verifies that valid XML fields assemble into correct UTC timestamp. [Fact] public void ParseTransitionTimestampUtc_WithValidXmlFields_AssemblesUtc() diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessSessionTests.cs new file mode 100644 index 0000000..0e61906 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessSessionTests.cs @@ -0,0 +1,591 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using Google.Protobuf.Collections; +using Google.Protobuf.WellKnownTypes; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Worker.MxAccess; +using ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport; + +namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess; + +/// +/// Direct tests of handle-lifetime cache +/// eviction and the phased ReadBulk snapshot path, driven through +/// with a recording fake +/// server. Nothing here needs an STA: the session methods are plain calls, +/// and ReadBulk's pump step is a test delegate that stands in for the COM +/// OnDataChange dispatch by writing the value cache directly. +/// +public sealed class MxAccessSessionTests +{ + private const string ReadBulkTimeoutText = "timed out"; + + /// + /// RemoveItem must evict the handle pair's OnWriteComplete completion + /// entry as well as its last-value entry; otherwise an + /// AddItem -> Write -> RemoveItem churn grows the completion cache by one + /// entry per cycle for the life of the worker. + /// + [Fact] + public void RemoveItem_EvictsValueAndWriteCompletionEntriesForTheHandle() + { + RecordingMxAccessServer server = new(); + CompletionCacheSink sink = new(); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache); + + int serverHandle = session.Register("client"); + int keep = session.AddItem(serverHandle, "Tag.Keep"); + int drop = session.AddItem(serverHandle, "Tag.Drop"); + RecordCompletion(sink.WriteCompletionCache, serverHandle, keep); + RecordCompletion(sink.WriteCompletionCache, serverHandle, drop); + valueCache.Set(serverHandle, keep, BuildDataChange(serverHandle, keep, 1)); + valueCache.Set(serverHandle, drop, BuildDataChange(serverHandle, drop, 2)); + + session.RemoveItem(serverHandle, drop); + + Assert.Equal(1, sink.WriteCompletionCache.Count); + Assert.Equal(0UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, drop)); + Assert.Equal(1UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, keep)); + Assert.False(valueCache.TryGet(serverHandle, drop, out _)); + Assert.True(valueCache.TryGet(serverHandle, keep, out _)); + } + + /// + /// A long-lived writer that keeps adding, writing, and removing items + /// must leave the completion cache empty — the leak this fix closes. + /// + [Fact] + public void AddItemWriteRemoveItemChurn_DoesNotGrowWriteCompletionCache() + { + RecordingMxAccessServer server = new(); + CompletionCacheSink sink = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink); + int serverHandle = session.Register("client"); + + for (int cycle = 0; cycle < 50; cycle++) + { + int itemHandle = session.AddItem(serverHandle, "Tag.Churn"); + session.Write(serverHandle, itemHandle, cycle, userId: 0); + + // Stand in for the sink's OnWriteComplete post-publish hook. + RecordCompletion(sink.WriteCompletionCache, serverHandle, itemHandle); + session.RemoveItem(serverHandle, itemHandle); + } + + Assert.Equal(0, sink.WriteCompletionCache.Count); + } + + /// + /// Unregister drops every item under the server in one MXAccess call, + /// so the per-item eviction never runs for them: both caches must be + /// evicted by server handle, and other servers' entries kept. + /// + [Fact] + public void Unregister_EvictsValueAndWriteCompletionEntriesForThatServerOnly() + { + RecordingMxAccessServer server = new(); + CompletionCacheSink sink = new(); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache); + + int doomed = session.Register("doomed"); + int survivor = session.Register("survivor"); + int doomedItemA = session.AddItem(doomed, "Tag.A"); + int doomedItemB = session.AddItem(doomed, "Tag.B"); + int survivorItem = session.AddItem(survivor, "Tag.C"); + RecordCompletion(sink.WriteCompletionCache, doomed, doomedItemA); + RecordCompletion(sink.WriteCompletionCache, doomed, doomedItemB); + RecordCompletion(sink.WriteCompletionCache, survivor, survivorItem); + valueCache.Set(doomed, doomedItemA, BuildDataChange(doomed, doomedItemA, 1)); + valueCache.Set(doomed, doomedItemB, BuildDataChange(doomed, doomedItemB, 2)); + valueCache.Set(survivor, survivorItem, BuildDataChange(survivor, survivorItem, 3)); + + session.Unregister(doomed); + + Assert.Equal(1, sink.WriteCompletionCache.Count); + Assert.Equal(1UL, sink.WriteCompletionCache.CurrentVersion(survivor, survivorItem)); + Assert.Equal(1, valueCache.Count); + Assert.True(valueCache.TryGet(survivor, survivorItem, out _)); + Assert.False(valueCache.TryGet(doomed, doomedItemA, out _)); + Assert.False(valueCache.TryGet(doomed, doomedItemB, out _)); + } + + /// + /// N uncached tags that never deliver an OnDataChange must cost one + /// shared timeout, not N x timeout, while each still reports its own + /// timed-out failure with its own item handle. + /// + [Fact] + public void ReadBulk_UncachedTagsThatNeverDeliver_WaitOneSharedTimeoutAndEachTimesOut() + { + RecordingMxAccessServer server = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink()); + int serverHandle = session.Register("client"); + string[] tags = { "Tag.1", "Tag.2", "Tag.3", "Tag.4", "Tag.5" }; + TimeSpan timeout = TimeSpan.FromMilliseconds(300); + + Stopwatch stopwatch = Stopwatch.StartNew(); + IReadOnlyList results = session.ReadBulk(serverHandle, tags, timeout, () => { }); + stopwatch.Stop(); + + // Serial would be >= 5 x 300 ms = 1500 ms. Allow generous slack for a + // loaded CI host while still failing the serial shape. + Assert.True( + stopwatch.Elapsed < TimeSpan.FromMilliseconds(1200), + $"ReadBulk took {stopwatch.Elapsed.TotalMilliseconds:F0} ms; expected one shared {timeout.TotalMilliseconds:F0} ms wait."); + Assert.True(stopwatch.Elapsed >= TimeSpan.FromMilliseconds(250)); + + Assert.Equal(tags.Length, results.Count); + HashSet handles = new(); + for (int index = 0; index < tags.Length; index++) + { + BulkReadResult result = results[index]; + Assert.Equal(tags[index], result.TagAddress); + Assert.False(result.WasSuccessful); + Assert.False(result.WasCached); + Assert.Equal(serverHandle, result.ServerHandle); + Assert.Equal(server.HandleForTag(tags[index]), result.ItemHandle); + Assert.Equal( + $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.", + result.ErrorMessage); + Assert.True(handles.Add(result.ItemHandle)); + } + + // Every snapshot handle is still torn down. + foreach (int handle in handles) + { + Assert.Contains($"UnAdvise:{serverHandle}:{handle}", server.Operations); + Assert.Contains($"RemoveItem:{serverHandle}:{handle}", server.Operations); + } + + Assert.Empty(session.HandleRegistry.ItemHandles); + } + + /// + /// The phased lifecycle: every uncached tag is added and advised before + /// the wait, and none is torn down until the wait is over. + /// + [Fact] + public void ReadBulk_AddsAndAdvisesEverySnapshotTagBeforeTearingAnyDown() + { + RecordingMxAccessServer server = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink()); + int serverHandle = session.Register("client"); + + session.ReadBulk(serverHandle, new[] { "Tag.A", "Tag.B", "Tag.C" }, TimeSpan.FromMilliseconds(50), () => { }); + + List lifecycle = server.Operations.Where(op => !op.StartsWith("Register", StringComparison.Ordinal)).ToList(); + int lastAdvise = lifecycle.FindLastIndex(op => op.StartsWith("Advise:", StringComparison.Ordinal)); + int firstUnAdvise = lifecycle.FindIndex(op => op.StartsWith("UnAdvise:", StringComparison.Ordinal)); + int firstRemove = lifecycle.FindIndex(op => op.StartsWith("RemoveItem:", StringComparison.Ordinal)); + Assert.Equal(3, lifecycle.Count(op => op.StartsWith("AddItem:", StringComparison.Ordinal))); + Assert.Equal(3, lifecycle.Count(op => op.StartsWith("Advise:", StringComparison.Ordinal))); + Assert.True(lastAdvise < firstUnAdvise, string.Join(",", lifecycle)); + Assert.True(lastAdvise < firstRemove, string.Join(",", lifecycle)); + } + + /// + /// Every per-tag outcome keeps its exact pre-batching shape and request + /// order in one mixed request: blank address, cached-and-advised fast + /// path, a tag that delivers, an AddItem failure, an Advise failure, and + /// a tag that never delivers. A delivering tag returns as soon as its + /// value lands rather than waiting out the timeout for the others. + /// + [Fact] + public void ReadBulk_MixedOutcomes_PreservesPerTagResultsAndOrder() + { + RecordingMxAccessServer server = new(); + server.FailAddItemFor.Add("Tag.AddFails"); + server.FailAdviseFor.Add("Tag.AdviseFails"); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache); + int serverHandle = session.Register("client"); + + // Caller-owned advised tag with a cached value: served from the cache. + int cachedHandle = session.AddItem(serverHandle, "Tag.Cached"); + session.Advise(serverHandle, cachedHandle); + valueCache.Set(serverHandle, cachedHandle, BuildDataChange(serverHandle, cachedHandle, 11)); + int operationsBeforeRead = server.Operations.Count; + + // Pump step: deliver Tag.Delivers' first OnDataChange once advised. + Action pumpStep = () => + { + if (server.TryGetAdvisedHandle("Tag.Delivers", out int handle) + && !valueCache.TryGet(serverHandle, handle, out _)) + { + valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 42)); + } + }; + + string[] tags = { " ", "Tag.Cached", "Tag.Delivers", "Tag.AddFails", "Tag.AdviseFails", "Tag.Silent" }; + IReadOnlyList results = session.ReadBulk(serverHandle, tags, TimeSpan.FromMilliseconds(150), pumpStep); + + Assert.Equal(tags.Length, results.Count); + for (int index = 0; index < tags.Length; index++) + { + Assert.Equal(tags[index], results[index].TagAddress); + } + + Assert.False(results[0].WasSuccessful); + Assert.Equal(0, results[0].ItemHandle); + Assert.Equal("Tag address is required.", results[0].ErrorMessage); + + Assert.True(results[1].WasSuccessful); + Assert.True(results[1].WasCached); + Assert.Equal(cachedHandle, results[1].ItemHandle); + Assert.Equal(11, results[1].Value.Int32Value); + + Assert.True(results[2].WasSuccessful); + Assert.False(results[2].WasCached); + Assert.Equal(server.HandleForTag("Tag.Delivers"), results[2].ItemHandle); + Assert.Equal(42, results[2].Value.Int32Value); + Assert.Equal(192, results[2].Quality); + + Assert.False(results[3].WasSuccessful); + Assert.Equal(0, results[3].ItemHandle); + Assert.Equal("AddItem failed for Tag.AddFails", results[3].ErrorMessage); + + Assert.False(results[4].WasSuccessful); + Assert.Equal(server.HandleForTag("Tag.AdviseFails"), results[4].ItemHandle); + Assert.Equal("Advise failed for Tag.AdviseFails", results[4].ErrorMessage); + + Assert.False(results[5].WasSuccessful); + Assert.Equal(server.HandleForTag("Tag.Silent"), results[5].ItemHandle); + Assert.Contains(ReadBulkTimeoutText, results[5].ErrorMessage, StringComparison.OrdinalIgnoreCase); + + // The caller's subscription is untouched; the snapshot handles are gone + // (the Advise-failed one removed without an UnAdvise). + List readOperations = server.Operations.Skip(operationsBeforeRead).ToList(); + Assert.DoesNotContain($"UnAdvise:{serverHandle}:{cachedHandle}", readOperations); + Assert.DoesNotContain($"RemoveItem:{serverHandle}:{cachedHandle}", readOperations); + Assert.DoesNotContain($"UnAdvise:{serverHandle}:{server.HandleForTag("Tag.AdviseFails")}", readOperations); + Assert.Contains($"RemoveItem:{serverHandle}:{server.HandleForTag("Tag.AdviseFails")}", readOperations); + Assert.Contains($"UnAdvise:{serverHandle}:{server.HandleForTag("Tag.Delivers")}", readOperations); + Assert.Contains($"RemoveItem:{serverHandle}:{server.HandleForTag("Tag.Silent")}", readOperations); + RegisteredItemHandle remaining = Assert.Single(session.HandleRegistry.ItemHandles); + Assert.Equal(cachedHandle, remaining.ItemHandle); + } + + /// + /// When every snapshot tag delivers, the wait ends as soon as the last + /// value lands instead of running out the deadline. + /// + [Fact] + public void ReadBulk_AllTagsDeliver_ReturnsWithoutWaitingOutTheTimeout() + { + RecordingMxAccessServer server = new(); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache); + int serverHandle = session.Register("client"); + string[] tags = { "Tag.1", "Tag.2", "Tag.3" }; + + Action pumpStep = () => + { + foreach (string tag in tags) + { + if (server.TryGetAdvisedHandle(tag, out int handle) && !valueCache.TryGet(serverHandle, handle, out _)) + { + valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, handle)); + } + } + }; + + Stopwatch stopwatch = Stopwatch.StartNew(); + IReadOnlyList results = session.ReadBulk(serverHandle, tags, TimeSpan.FromSeconds(10), pumpStep); + stopwatch.Stop(); + + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(5), $"ReadBulk took {stopwatch.Elapsed.TotalMilliseconds:F0} ms."); + Assert.All(results, result => Assert.True(result.WasSuccessful)); + for (int index = 0; index < tags.Length; index++) + { + Assert.Equal(results[index].ItemHandle, results[index].Value.Int32Value); + } + } + + /// + /// Ghost entry: an OnDataChange/OnWriteComplete already queued when + /// RemoveItem ran is dispatched after the eviction and recreates the + /// entries. When MXAccess later reuses that item id, AddItem must drop + /// them so ReadBulk's cached path cannot serve the stale value. + /// + [Fact] + public void AddItem_ReusingAnEvictedHandleId_DropsGhostCacheEntries() + { + RecordingMxAccessServer server = new(); + CompletionCacheSink sink = new(); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache); + int serverHandle = session.Register("client"); + + int first = session.AddItem(serverHandle, "Tag.Old"); + session.Advise(serverHandle, first); + session.RemoveItem(serverHandle, first); + + // Late dispatch of callbacks queued before the RemoveItem. + valueCache.Set(serverHandle, first, BuildDataChange(serverHandle, first, 999)); + RecordCompletion(sink.WriteCompletionCache, serverHandle, first); + + server.NextItemHandleOverride = first; + int reused = session.AddItem(serverHandle, "Tag.New"); + Assert.Equal(first, reused); + + Assert.False(valueCache.TryGet(serverHandle, reused, out _)); + Assert.Equal(0UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, reused)); + + // Once advised, ReadBulk on the new tag does not report the ghost as cached. + session.Advise(serverHandle, reused); + BulkReadResult result = Assert.Single(session.ReadBulk(serverHandle, new[] { "Tag.New" }, TimeSpan.FromMilliseconds(30), () => { })); + Assert.False(result.WasCached); + Assert.False(result.WasSuccessful); + } + + /// + /// A handle MXAccess returns while the registry still tracks it is a + /// live item, so its cached value must survive the AddItem. + /// + [Fact] + public void AddItem_ReturningAStillRegisteredHandle_KeepsItsCachedValue() + { + RecordingMxAccessServer server = new() { ReuseHandleForSameTag = true }; + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache); + int serverHandle = session.Register("client"); + int handle = session.AddItem(serverHandle, "Tag.Live"); + valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 5)); + + Assert.Equal(handle, session.AddItem(serverHandle, "Tag.Live")); + + Assert.True(valueCache.TryGet(serverHandle, handle, out _)); + } + + /// + /// Duplicate addresses that MXAccess resolves to one handle: the handle + /// is advised, unadvised, and removed exactly once, and every duplicate + /// receives the shared outcome. + /// + [Fact] + public void ReadBulk_DuplicateTagsResolvingToOneHandle_AdviseAndTearDownOnceAndFanOutResult() + { + RecordingMxAccessServer server = new() { ReuseHandleForSameTag = true }; + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache); + int serverHandle = session.Register("client"); + + Action pumpStep = () => + { + if (server.TryGetAdvisedHandle("Tag.Dup", out int handle) && !valueCache.TryGet(serverHandle, handle, out _)) + { + valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 77)); + } + }; + + IReadOnlyList results = session.ReadBulk( + serverHandle, + new[] { "Tag.Dup", "Tag.Other", "Tag.Dup" }, + TimeSpan.FromMilliseconds(100), + pumpStep); + + int dupHandle = server.HandleForTag("Tag.Dup"); + string advise = "Advise:" + serverHandle + ":" + dupHandle; + string unAdvise = "UnAdvise:" + serverHandle + ":" + dupHandle; + string remove = "RemoveItem:" + serverHandle + ":" + dupHandle; + Assert.True(results[0].WasSuccessful); + Assert.True(results[2].WasSuccessful); + Assert.Equal(dupHandle, results[0].ItemHandle); + Assert.Equal(dupHandle, results[2].ItemHandle); + Assert.Equal(77, results[2].Value.Int32Value); + Assert.Equal("Tag.Dup", results[2].TagAddress); + Assert.False(results[1].WasSuccessful); + Assert.Equal("Tag.Other", results[1].TagAddress); + + Assert.Equal(1, server.Operations.Count(op => op == advise)); + Assert.Equal(1, server.Operations.Count(op => op == unAdvise)); + Assert.Equal(1, server.Operations.Count(op => op == remove)); + Assert.Empty(session.HandleRegistry.ItemHandles); + } + + /// + /// Large batches pump during the add/advise and teardown phases too, + /// not only inside the wait, so the STA heartbeat stays fresh. + /// + [Fact] + public void ReadBulk_LargeBatch_PumpsDuringAddAdviseAndTeardownPhases() + { + RecordingMxAccessServer server = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink()); + int serverHandle = session.Register("client"); + string[] tags = Enumerable.Range(0, 3 * MxAccessSession.ReadBulkPumpInterval).Select(i => "Tag." + i).ToArray(); + List pumpAtOperation = new(); + + session.ReadBulk(serverHandle, tags, TimeSpan.FromMilliseconds(20), () => pumpAtOperation.Add(server.Operations.Count)); + + int lastAdvise = server.Operations.FindLastIndex(op => op.StartsWith("Advise:", StringComparison.Ordinal)); + int firstUnAdvise = server.Operations.FindIndex(op => op.StartsWith("UnAdvise:", StringComparison.Ordinal)); + Assert.Contains(pumpAtOperation, at => at < lastAdvise); + Assert.Contains(pumpAtOperation, at => at > firstUnAdvise); + } + + private static void RecordCompletion(MxAccessWriteCompletionCache cache, int serverHandle, int itemHandle) + { + RepeatedField statuses = new() { new MxStatusProxy { Category = MxStatusCategory.Ok } }; + cache.Record(serverHandle, itemHandle, statuses); + } + + private static MxEvent BuildDataChange(int serverHandle, int itemHandle, int value) + { + MxEvent mxEvent = new() + { + Family = MxEventFamily.OnDataChange, + ServerHandle = serverHandle, + ItemHandle = itemHandle, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(new DateTime(2026, 10, 22, 0, 0, 0, DateTimeKind.Utc)), + Value = new MxValue + { + DataType = MxDataType.Integer, + VariantType = "VT_I4", + Int32Value = value, + }, + OnDataChange = new OnDataChangeEvent(), + }; + mxEvent.Statuses.Add(new MxStatusProxy { Category = MxStatusCategory.Ok }); + return mxEvent; + } + + /// Test sink that owns a real completion cache (Attach/Detach are no-ops). + private sealed class CompletionCacheSink : IMxAccessEventSink, IWriteCompletionCacheProvider + { + /// + public MxAccessWriteCompletionCache WriteCompletionCache { get; } = new(); + + /// + public void Attach(object mxAccessComObject, string sessionId) + { + } + + /// + public void Detach() + { + } + } + + /// + /// Fake server that hands out increasing handles, records every + /// lifecycle call, and can fail AddItem or Advise for named tags. + /// + private sealed class RecordingMxAccessServer : IMxAccessServer + { + private readonly NoopMxAccessServer noop = new(); + private readonly Dictionary handleByTag = new(StringComparer.Ordinal); + private readonly Dictionary tagByHandle = new(); + private readonly HashSet advised = new(); + private int nextServerHandle = 10; + private int nextItemHandle = 100; + + public List Operations { get; } = new(); + + public HashSet FailAddItemFor { get; } = new(StringComparer.Ordinal); + + public HashSet FailAdviseFor { get; } = new(StringComparer.Ordinal); + + /// When set, the next AddItem returns this handle (simulates MXAccess reusing an id). + public int? NextItemHandleOverride { get; set; } + + /// When true, AddItem of an already-added address returns the existing handle. + public bool ReuseHandleForSameTag { get; set; } + + public int HandleForTag(string tag) => handleByTag[tag]; + + public bool TryGetAdvisedHandle(string tag, out int handle) + { + return handleByTag.TryGetValue(tag, out handle) && advised.Contains(handle); + } + + public int Register(string clientName) + { + Operations.Add($"Register:{clientName}"); + return nextServerHandle++; + } + + public void Unregister(int serverHandle) => Operations.Add($"Unregister:{serverHandle}"); + + public int AddItem(int serverHandle, string itemDefinition) + { + Operations.Add($"AddItem:{serverHandle}:{itemDefinition}"); + if (FailAddItemFor.Contains(itemDefinition)) + { + throw new InvalidOperationException($"AddItem failed for {itemDefinition}"); + } + + int handle; + if (NextItemHandleOverride is int forced) + { + NextItemHandleOverride = null; + handle = forced; + } + else if (ReuseHandleForSameTag && handleByTag.TryGetValue(itemDefinition, out int existing)) + { + handle = existing; + } + else + { + handle = nextItemHandle++; + } + + handleByTag[itemDefinition] = handle; + tagByHandle[handle] = itemDefinition; + return handle; + } + + public int AddItem2(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition); + + public void RemoveItem(int serverHandle, int itemHandle) => Operations.Add($"RemoveItem:{serverHandle}:{itemHandle}"); + + public void Advise(int serverHandle, int itemHandle) + { + Operations.Add($"Advise:{serverHandle}:{itemHandle}"); + if (tagByHandle.TryGetValue(itemHandle, out string? tag) && FailAdviseFor.Contains(tag)) + { + throw new InvalidOperationException($"Advise failed for {tag}"); + } + + advised.Add(itemHandle); + } + + public void UnAdvise(int serverHandle, int itemHandle) + { + Operations.Add($"UnAdvise:{serverHandle}:{itemHandle}"); + advised.Remove(itemHandle); + } + + public void AdviseSupervisory(int serverHandle, int itemHandle) => Advise(serverHandle, itemHandle); + + public int AddBufferedItem(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition); + + public void SetBufferedUpdateInterval(int serverHandle, int updateIntervalMilliseconds) => + noop.SetBufferedUpdateInterval(serverHandle, updateIntervalMilliseconds); + + public object Suspend(int serverHandle, int itemHandle) => noop.Suspend(serverHandle, itemHandle); + + public object Activate(int serverHandle, int itemHandle) => noop.Activate(serverHandle, itemHandle); + + public void Write(int serverHandle, int itemHandle, object? value, int userId) => + Operations.Add($"Write:{serverHandle}:{itemHandle}"); + + public void Write2(int serverHandle, int itemHandle, object? value, object? timestampValue, int userId) => + Operations.Add($"Write2:{serverHandle}:{itemHandle}"); + + public void WriteSecured(int serverHandle, int itemHandle, int currentUserId, int verifierUserId, object? value) => + Operations.Add($"WriteSecured:{serverHandle}:{itemHandle}"); + + public void WriteSecured2(int serverHandle, int itemHandle, int currentUserId, int verifierUserId, object? value, object? timestampValue) => + Operations.Add($"WriteSecured2:{serverHandle}:{itemHandle}"); + + public int AuthenticateUser(int serverHandle, string verifyUser, string verifyUserPassword) => 0; + + public int ArchestrAUserToId(int serverHandle, string userIdGuid) => 0; + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs index 07d378d..1294798 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Runtime.InteropServices; using System.Threading; @@ -573,6 +574,181 @@ public sealed class MxAccessStaSessionTests "Expected StaCallInProgress to clear once the alarm poll returned."); } + /// + /// A transient PollOnce failure must not fault the whole session: with a + /// failure threshold of 3, a poll pattern that never fails three times in + /// a row (fail, fail, succeed, ...) keeps the session healthy, and each + /// transient failure is logged with type/HRESULT only — never the + /// exception message. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAlarmPollLoop_TransientFailuresBelowThreshold_DoNotFaultSession() + { + FakeAlarmCommandHandler handler = new() + { + PollExceptionFactory = call => call % 3 == 0 + ? null + : new COMException("secret-tag-name GetXmlCurrentAlarms2 failed.", unchecked((int)0x80004005)), + }; + ConcurrentQueue warnings = new(); + using StaRuntime runtime = CreateRuntime(); + MxAccessEventQueue eventQueue = new(); + using MxAccessStaSession session = new( + runtime, + new FakeMxAccessComObjectFactory(), + new FakeMxAccessEventSink(), + eventQueue, + (_eq, _affinity, _comFactory) => handler); + session.AlarmPollInterval = TimeSpan.FromMilliseconds(10); + session.AlarmPollFailureThreshold = 3; + session.AlarmPollWarningSink = warnings.Enqueue; + + await session.StartAsync("session-1", workerProcessId: 1); + + DateTime deadline = DateTime.UtcNow.AddSeconds(10); + while (handler.PollCount < 12 && !eventQueue.IsFaulted && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.False(eventQueue.IsFaulted, "Transient alarm-poll failures below the threshold must not fault the session."); + Assert.True(handler.PollCount >= 12, $"Expected the poll loop to keep polling; saw {handler.PollCount} polls."); + Assert.NotEmpty(warnings); + Assert.All(warnings, line => + { + Assert.Contains("event=AlarmPollFailed", line, StringComparison.Ordinal); + Assert.Contains("hresult=0x80004005", line, StringComparison.Ordinal); + Assert.DoesNotContain("secret-tag-name", line, StringComparison.Ordinal); + }); + } + + /// + /// A run of failures reaching the threshold with no success in between + /// faults the session exactly once and stops the loop: two warnings, then + /// the fault on the third consecutive failure, then no further polls. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAlarmPollLoop_ConsecutiveFailuresReachThreshold_FaultsAndStops() + { + FakeAlarmCommandHandler handler = new() + { + PollException = new COMException("GetXmlCurrentAlarms2 failed.", unchecked((int)0x80004005)), + }; + ConcurrentQueue warnings = new(); + using StaRuntime runtime = CreateRuntime(); + MxAccessEventQueue eventQueue = new(); + using MxAccessStaSession session = new( + runtime, + new FakeMxAccessComObjectFactory(), + new FakeMxAccessEventSink(), + eventQueue, + (_eq, _affinity, _comFactory) => handler); + session.AlarmPollInterval = TimeSpan.FromMilliseconds(10); + session.AlarmPollFailureThreshold = 3; + session.AlarmPollWarningSink = warnings.Enqueue; + + await session.StartAsync("session-1", workerProcessId: 1); + + DateTime deadline = DateTime.UtcNow.AddSeconds(10); + while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.True(eventQueue.IsFaulted); + await Task.Delay(200); + Assert.Equal(3, handler.PollCount); + Assert.Equal(2, warnings.Count); + WorkerFault? fault = session.DrainFault(); + Assert.NotNull(fault); + Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, fault!.Category); + } + + /// + /// WRK-02: when the STA loop dies after the session started, the death is + /// recorded as a StaHung fault on the event queue — the path the IPC drain + /// loop turns into a WorkerFault frame and a terminated pipe session. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StaLoopDeath_AfterStart_RecordsStaHungFaultOnEventQueue() + { + ControllableMessagePump pump = new(); + using StaRuntime runtime = new(new NoopComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25)); + MxAccessEventQueue eventQueue = new(); + using MxAccessStaSession session = new( + runtime, + new FakeMxAccessComObjectFactory(), + new FakeMxAccessEventSink(), + eventQueue); + + await session.StartAsync("session-1", workerProcessId: 1); + Assert.False(eventQueue.IsFaulted); + + pump.ThrowOnNextPump(new InvalidOperationException("secret-tag-name pump failed.")); + + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.True(eventQueue.IsFaulted, "Expected the STA loop death to fault the event queue."); + WorkerFault? fault = session.DrainFault(); + Assert.NotNull(fault); + Assert.Equal(WorkerFaultCategory.StaHung, fault!.Category); + Assert.Equal(typeof(InvalidOperationException).FullName, fault.ExceptionType); + Assert.Contains("STA loop terminated", fault.DiagnosticMessage, StringComparison.Ordinal); + Assert.Contains("exceptionType=System.InvalidOperationException", fault.DiagnosticMessage, StringComparison.Ordinal); + Assert.Contains("hresult=0x", fault.DiagnosticMessage, StringComparison.Ordinal); + Assert.DoesNotContain("secret-tag-name", fault.DiagnosticMessage, StringComparison.Ordinal); + Assert.DoesNotContain("secret-tag-name", fault.ProtocolStatus.Message, StringComparison.Ordinal); + Assert.Equal(ProtocolStatusCode.WorkerUnavailable, fault.ProtocolStatus.Code); + } + + /// + /// After the STA loop dies, the alarm poll's fast-failed InvokeAsync must + /// not record a second, mislabelled alarm-poll fault: once the StaHung + /// fault is drained, nothing else is recorded. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAlarmPollLoop_AfterStaLoopFault_DoesNotRecordAlarmPollFault() + { + ControllableMessagePump pump = new(); + FakeAlarmCommandHandler handler = new(); + using StaRuntime runtime = new(new NoopComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25)); + MxAccessEventQueue eventQueue = new(); + using MxAccessStaSession session = new( + runtime, + new FakeMxAccessComObjectFactory(), + new FakeMxAccessEventSink(), + eventQueue, + (_eq, _affinity, _comFactory) => handler); + session.AlarmPollInterval = TimeSpan.FromMilliseconds(10); + session.AlarmPollFailureThreshold = 1; + session.AlarmPollWarningSink = _ => { }; + + await session.StartAsync("session-1", workerProcessId: 1); + pump.ThrowOnNextPump(new InvalidOperationException("pump failed")); + + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + WorkerFault? first = session.DrainFault(); + Assert.NotNull(first); + Assert.Equal(WorkerFaultCategory.StaHung, first!.Category); + + // Give the poll loop several ticks to (wrongly) record its own fault. + await Task.Delay(300); + Assert.Null(session.DrainFault()); + } + /// /// The STA-affinity guard throws when an /// IMxAccessAlarmConsumer call is attempted off the thread that created @@ -621,6 +797,9 @@ public sealed class MxAccessStaSessionTests /// Exception thrown by PollOnce; null to succeed. public Exception? PollException { get; set; } + /// Per-call exception factory (1-based call number); null result succeeds. + public Func? PollExceptionFactory { get; set; } + /// When set, blocks until is called. public bool BlockPoll { get; set; } @@ -699,6 +878,22 @@ public sealed class MxAccessStaSessionTests { throw PollException; } + + Func? factory = PollExceptionFactory; + if (factory is not null) + { + int callNumber; + lock (gate) + { + callNumber = pollCount; + } + + Exception? exception = factory(callNumber); + if (exception is not null) + { + throw exception; + } + } } /// 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 7173434..1a336f9 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs @@ -2,6 +2,7 @@ using System; using System.Threading; using System.Threading.Tasks; using ZB.MOM.WW.MxGateway.Worker.Sta; +using ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport; namespace ZB.MOM.WW.MxGateway.Worker.Tests.Sta; @@ -155,6 +156,145 @@ public sealed class StaRuntimeTests Assert.Contains("shutting down", exception.Message); } + /// + /// A backlog of queued work items must not starve inbound COM + /// callbacks: the loop pumps Windows messages between items rather + /// than only after draining the whole queue. A gate item holds the STA + /// while two more items queue behind it; once released, the second + /// queued item must observe a pump that happened after the first. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ProcessQueuedCommands_PumpsMessagesBetweenQueuedWorkItems() + { + ControllableMessagePump pump = new(); + using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromSeconds(30)); + runtime.Start(); + + using ManualResetEventSlim gateEntered = new(false); + using ManualResetEventSlim releaseGate = new(false); + Task gate = runtime.InvokeAsync(() => + { + gateEntered.Set(); + releaseGate.Wait(TimeSpan.FromSeconds(10)); + }); + Assert.True(gateEntered.Wait(TimeSpan.FromSeconds(5))); + + Task first = runtime.InvokeAsync(() => pump.PumpCount); + Task second = runtime.InvokeAsync(() => pump.PumpCount); + releaseGate.Set(); + + await gate; + int pumpsSeenByFirst = await first; + int pumpsSeenBySecond = await second; + + Assert.True( + pumpsSeenBySecond > pumpsSeenByFirst, + $"Expected a pump between queued work items (first saw {pumpsSeenByFirst}, second saw {pumpsSeenBySecond})."); + } + + /// + /// WRK-02: an exception that kills the STA loop after a successful + /// must be surfaced — recorded on + /// , handed to the registered fault + /// handler, and turned into fast failures for every later + /// — + /// rather than parked in a startup field nothing reads. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task LoopException_AfterStart_IsReportedToHandlerAndFailsLaterCommands() + { + ControllableMessagePump pump = new(); + RecordingComApartmentInitializer initializer = new(); + using StaRuntime runtime = new(initializer, pump, TimeSpan.FromMilliseconds(25)); + runtime.Start(); + + TaskCompletionSource reported = new(TaskCreationOptions.RunContinuationsAsynchronously); + runtime.RegisterLoopFaultHandler(exception => reported.TrySetResult(exception)); + Assert.Null(runtime.LoopFault); + + InvalidOperationException loopException = new("pump failed"); + pump.ThrowOnNextPump(loopException); + + Task completed = await Task.WhenAny(reported.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.Same(reported.Task, completed); + Assert.Same(loopException, await reported.Task); + Assert.Same(loopException, runtime.LoopFault); + + InvalidOperationException rejected = await Assert.ThrowsAsync( + () => runtime.InvokeAsync(() => 1)); + Assert.IsNotType(rejected); + Assert.Same(loopException, rejected.InnerException); + + // The dying thread still runs its apartment cleanup. + Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5))); + Assert.False(runtime.IsRunning); + Assert.Equal(1, initializer.UninitializeCount); + } + + /// + /// A handler registered after the loop already died still receives the + /// fault, so the window between Start and registration loses nothing. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RegisterLoopFaultHandler_AfterLoopDied_InvokesHandlerImmediately() + { + ControllableMessagePump pump = new(); + using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25)); + runtime.Start(); + InvalidOperationException loopException = new("pump failed"); + pump.ThrowOnNextPump(loopException); + + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (runtime.LoopFault is null && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Exception? seen = null; + runtime.RegisterLoopFaultHandler(exception => seen = exception); + + Assert.Same(loopException, seen); + } + + /// A requested shutdown is not a loop fault. + [Fact] + public void Shutdown_DoesNotRecordLoopFault() + { + using StaRuntime runtime = new(new RecordingComApartmentInitializer(), new ControllableMessagePump(), TimeSpan.FromMilliseconds(25)); + runtime.Start(); + bool invoked = false; + runtime.RegisterLoopFaultHandler(_ => invoked = true); + + Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5))); + + Assert.Null(runtime.LoopFault); + Assert.False(invoked); + } + + /// + /// A loop that throws while a requested shutdown is in progress is + /// being torn down on purpose: the loop-fault handler must not run. + /// + [Fact] + public void LoopException_DuringRequestedShutdown_DoesNotInvokeHandler() + { + ControllableMessagePump pump = new(); + using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromSeconds(30)); + runtime.Start(); + bool invoked = false; + runtime.RegisterLoopFaultHandler(_ => invoked = true); + + // The loop is parked in the 30 s idle wait; Shutdown wakes it and the + // very next pump throws with shutdownRequested already set. + pump.ThrowOnNextPump(new InvalidOperationException("pump failed during shutdown")); + Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5))); + + Assert.False(invoked); + } + private static StaRuntime CreateRuntime(RecordingComApartmentInitializer initializer) { return new StaRuntime( diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/ControllableMessagePump.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/ControllableMessagePump.cs new file mode 100644 index 0000000..7658629 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/ControllableMessagePump.cs @@ -0,0 +1,47 @@ +using System; +using System.Threading; +using ZB.MOM.WW.MxGateway.Worker.Sta; + +namespace ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport; + +/// +/// Test that counts pump calls and can be armed +/// to throw from the next pump, so tests can observe the STA loop's pump +/// interleaving and drive a post-start loop death without a real Win32 +/// message queue. The idle wait blocks on the command wake event exactly +/// like the production pump's MsgWaitForMultipleObjectsEx does. +/// +internal sealed class ControllableMessagePump : IStaMessagePump +{ + private int pumpCount; + private Exception? throwOnNextPump; + + /// Gets the number of calls so far. + public int PumpCount => Volatile.Read(ref pumpCount); + + /// Arms the pump to throw from its next pump call. + /// The exception to throw once. + public void ThrowOnNextPump(Exception exception) + { + Volatile.Write(ref throwOnNextPump, exception); + } + + /// + public void WaitForWorkOrMessages(WaitHandle commandWakeEvent, TimeSpan timeout) + { + commandWakeEvent.WaitOne(timeout); + } + + /// + public int PumpPendingMessages() + { + Interlocked.Increment(ref pumpCount); + Exception? exception = Interlocked.Exchange(ref throwOnNextPump, null); + if (exception is not null) + { + throw exception; + } + + return 0; + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Bootstrap/WorkerExitCode.cs b/src/ZB.MOM.WW.MxGateway.Worker/Bootstrap/WorkerExitCode.cs index 7045a38..416e6fa 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Bootstrap/WorkerExitCode.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Bootstrap/WorkerExitCode.cs @@ -9,4 +9,13 @@ public enum WorkerExitCode MissingNonce = 4, PipeConnectionFailed = 5, ProtocolViolation = 6, + + /// + /// An exception escaped to AppDomain.UnhandledException on some + /// thread (a background task, a COM callback thread, a timer) rather + /// than through WorkerApplication.Run's own catch blocks. The + /// worker logs it and exits with this code instead of the CLR's crash + /// exit code. + /// + UnhandledException = 7, } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs index 67273e9..25418bf 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs @@ -246,13 +246,32 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler { IMxAccessAlarmConsumer primary = consumerFactory() ?? throw new InvalidOperationException("Alarm consumer factory returned null."); - IMxAccessAlarmConsumer standby = BuildStandby(watchList); - AlarmFailoverConfig? failoverConfig = command.Failover; - FailoverSettings settings = new FailoverSettings( - failoverConfig?.ConsecutiveFailureThreshold ?? 3, - failoverConfig?.FailbackProbeIntervalSeconds ?? 30, - failoverConfig?.FailbackStableProbes ?? 3); - return new FailoverAlarmConsumer(primary, standby, settings); + IMxAccessAlarmConsumer? standby = null; + try + { + standby = BuildStandby(watchList); + AlarmFailoverConfig? failoverConfig = command.Failover; + FailoverSettings settings = new FailoverSettings( + failoverConfig?.ConsecutiveFailureThreshold ?? 3, + failoverConfig?.FailbackProbeIntervalSeconds ?? 30, + failoverConfig?.FailbackStableProbes ?? 3); + return new FailoverAlarmConsumer(primary, standby, settings); + } + catch + { + // The primary (and a standby that was built before the + // composite constructor threw) is owned by nothing yet: the + // failed Subscribe never reaches the dispatcher that would + // dispose it, so release it here or its wnwrap COM object + // leaks for the life of the worker. + if (standby is not null) + { + try { standby.Dispose(); } catch { /* swallow — preserve the build failure */ } + } + + try { primary.Dispose(); } catch { /* swallow — preserve the build failure */ } + throw; + } } // Alarmmgr, or Unspecified with an empty watch list — primary only. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmRecordTransitionMapper.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmRecordTransitionMapper.cs index aed471c..3fdd8e3 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmRecordTransitionMapper.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmRecordTransitionMapper.cs @@ -49,6 +49,17 @@ public static class AlarmRecordTransitionMapper /// previous=*Rtn + current=*Alm → Raise (re-trigger after clear). /// Anything else → Unspecified (no proto kind to emit). /// + /// + /// is never returned. It is + /// reserved in the contract, but neither alarm source gives the worker + /// a native re-alarm signal: the alarmmgr XML carries only the four + /// *_ALM/*_RTN states, so a return-to-alarm is + /// indistinguishable from a fresh raise and is reported as Raise, and + /// the subtag source observes no re-alarm counter. Deliberately not + /// synthesized (the gateway forwards only transitions the source + /// reports — see docs/AlarmClientDiscovery.md, "Retrigger + /// transitions"). + /// /// /// The previous alarm state kind. /// The current alarm state kind. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs index ee394ab..860f4ad 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs @@ -222,6 +222,13 @@ public sealed class MxAccessSession : IDisposable mxAccessServer.Unregister(serverHandle); handleRegistry.UnregisterServerHandle(serverHandle); + // Unregister drops every item under the server in one call, so the + // per-item evictions in RemoveItem never run for them. Evict both + // per-handle caches by server so a long-lived worker that registers + // and unregisters clients does not accumulate one entry per item it + // ever wrote or read, and a reused server handle id starts clean. + valueCache.RemoveServer(serverHandle); + writeCompletionCache.RemoveServer(serverHandle); } /// Adds an item to an MXAccess server and returns the item handle. @@ -235,6 +242,7 @@ public sealed class MxAccessSession : IDisposable ThrowIfDisposed(); int itemHandle = mxAccessServer.AddItem(serverHandle, itemDefinition); + EvictStaleHandleState(serverHandle, itemHandle); handleRegistry.RegisterItemHandle( serverHandle, itemHandle, @@ -258,6 +266,7 @@ public sealed class MxAccessSession : IDisposable ThrowIfDisposed(); int itemHandle = mxAccessServer.AddItem2(serverHandle, itemDefinition, itemContext); + EvictStaleHandleState(serverHandle, itemHandle); handleRegistry.RegisterItemHandle( serverHandle, itemHandle, @@ -268,6 +277,30 @@ public sealed class MxAccessSession : IDisposable return itemHandle; } + /// + /// Drops cached per-handle state left over from a previous lifetime of + /// a handle id MXAccess just handed out again. RemoveItem and Unregister + /// already evict, but an OnDataChange or OnWriteComplete that was + /// queued on the STA when they ran is dispatched afterwards (the loop + /// pumps between work items) and recreates the entry. Without this, a + /// reused id would let ReadBulk's cached fast path serve that stale + /// value as was_cached = true. A handle the registry still + /// tracks is a live item (MXAccess returned an existing handle), so its + /// current entries are kept. + /// + private void EvictStaleHandleState( + int serverHandle, + int itemHandle) + { + if (handleRegistry.ContainsItemHandle(serverHandle, itemHandle)) + { + return; + } + + valueCache.Remove(serverHandle, itemHandle); + writeCompletionCache.Remove(serverHandle, itemHandle); + } + /// Removes an item from an MXAccess server. /// Handle returned by the worker. /// Handle returned by the worker. @@ -283,6 +316,11 @@ public sealed class MxAccessSession : IDisposable // same handle id (which MXAccess may reuse) does not serve a stale // OnDataChange snapshot from the previous lifetime. valueCache.Remove(serverHandle, itemHandle); + // Same for the OnWriteComplete completion entry: the event sink records + // one per written handle pair, so without this an AddItem -> Write -> + // RemoveItem churn grows the cache by one entry per cycle for the life + // of the worker. + writeCompletionCache.Remove(serverHandle, itemHandle); } /// Advises on item changes with plain subscription. @@ -397,6 +435,7 @@ public sealed class MxAccessSession : IDisposable ThrowIfDisposed(); int itemHandle = mxAccessServer.AddBufferedItem(serverHandle, itemDefinition, itemContext); + EvictStaleHandleState(serverHandle, itemHandle); handleRegistry.RegisterItemHandle( serverHandle, itemHandle, @@ -870,14 +909,28 @@ public sealed class MxAccessSession : IDisposable /// OnDataChange value if the tag is already advised AND a cached value /// exists (no subscription side effects); otherwise takes the AddItem /// + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle itself. - /// bounds the wait per-tag in the snapshot - /// case; is invoked on every poll - /// iteration so the worker's STA can dispatch the incoming MXAccess - /// message that carries the value. + /// bounds the wait for every snapshot tag; + /// is invoked on every poll iteration so the + /// worker's STA can dispatch the incoming MXAccess message that carries + /// the value. /// + /// + /// The snapshot tags are processed in phases rather than one at a time: + /// every cache probe runs first, then every uncached tag is added and + /// advised, then one wait with a single shared deadline collects the + /// first OnDataChange per handle, then every snapshot handle is torn + /// down. A serial lifecycle held the STA for up to N x timeout when N + /// tags never produced a value; the phased form holds it for at most + /// one timeout. Each tag's result is still its own: a tag that never + /// delivered fails with the same timed-out diagnostic and its own item + /// handle, an AddItem or Advise failure fails only that tag, and results + /// keep request order. The cache probes run before any snapshot AddItem + /// so a later tag can never be served from an earlier tag's own + /// snapshot subscription. + /// /// The MXAccess server handle. /// The tag addresses to read. - /// The timeout per tag. + /// The timeout bounding the shared snapshot wait. /// Action invoked on each poll iteration. /// The per-tag read results, in the order the tags were provided. public IReadOnlyList ReadBulk( @@ -897,91 +950,225 @@ public sealed class MxAccessSession : IDisposable throw new ArgumentNullException(nameof(pumpStep)); } - List results = new(tagAddresses.Count); - foreach (string? tagAddress in tagAddresses) + int tagCount = tagAddresses.Count; + BulkReadResult?[] results = new BulkReadResult?[tagCount]; + List snapshots = new(); + + // Phase 1: validation and the cached-and-advised fast path. A tag + // already advised by the caller with a cached value is answered without + // touching the existing subscription — the caller didn't create it, so + // ReadBulk must not tear it down. + for (int index = 0; index < tagCount; index++) { + string? tagAddress = tagAddresses[index]; if (string.IsNullOrWhiteSpace(tagAddress)) { - results.Add(FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required.")); + results[index] = FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required."); continue; } - results.Add(ReadOneTag(serverHandle, tagAddress, timeout, pumpStep)); + if (TryGetCachedReadFor(serverHandle, tagAddress!, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue)) + { + results[index] = SucceededRead(serverHandle, tagAddress!, cachedItemHandle, wasCached: true, cachedValue); + continue; + } + + snapshots.Add(new SnapshotRead(index, tagAddress!)); } - return results; + if (snapshots.Count > 0) + { + try + { + ReadSnapshots(serverHandle, snapshots, results, timeout, pumpStep); + } + finally + { + TearDownSnapshots(serverHandle, snapshots, pumpStep); + } + } + + List ordered = new(tagCount); + for (int index = 0; index < tagCount; index++) + { + ordered.Add(results[index]!); + } + + return ordered; } - private BulkReadResult ReadOneTag( + /// + /// Number of snapshot items ReadBulk adds/advises (phase 2) or tears + /// down (phase 4) between pump steps. Those phases make one or two COM + /// calls per tag without waiting, so a large batch would otherwise hold + /// the STA with no pump and no activity mark for its whole length; a + /// pump every items keeps inbound + /// callbacks flowing and keeps the STA heartbeat fresh, so a big batch + /// cannot drift toward HeartbeatStuckCeiling. + /// + internal const int ReadBulkPumpInterval = 16; + + private void ReadSnapshots( int serverHandle, - string tagAddress, + List snapshots, + BulkReadResult?[] results, TimeSpan timeout, Action pumpStep) { - // 1. Cached-and-advised fast path: scan the registry for a live item - // handle matching this tag and check whether the value cache has a - // payload for it. If so, return the cached value without touching - // the existing subscription — the caller didn't create it, so - // ReadBulk must not tear it down. - if (TryGetCachedReadFor(serverHandle, tagAddress, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue)) + // Phase 2: reserve our own item handle and advise for every uncached + // tag. A per-tag AddItem/Advise failure fails only that tag. When + // MXAccess hands back a handle an earlier tag in this batch already + // advised (a duplicate address), that tag shares the earlier tag's + // subscription instead of advising the handle a second time, and gets + // the same outcome fanned out to it after the wait. + List waiting = new(snapshots.Count); + Dictionary advisedByHandle = new(); + List sharing = new(); + for (int position = 0; position < snapshots.Count; position++) { - return SucceededRead( - serverHandle, - tagAddress, - cachedItemHandle, - wasCached: true, - cachedValue); + SnapshotRead snapshot = snapshots[position]; + try + { + if (position > 0 && position % ReadBulkPumpInterval == 0) + { + pumpStep(); + } + + snapshot.ItemHandle = AddItem(serverHandle, snapshot.TagAddress); + if (advisedByHandle.TryGetValue(snapshot.ItemHandle, out SnapshotRead? owner)) + { + snapshot.SharedWith = owner; + sharing.Add(snapshot); + continue; + } + + snapshot.Baseline = valueCache.CurrentVersion(serverHandle, snapshot.ItemHandle); + Advise(serverHandle, snapshot.ItemHandle); + snapshot.Advised = true; + advisedByHandle[snapshot.ItemHandle] = snapshot; + waiting.Add(snapshot); + } + catch (Exception exception) + { + results[snapshot.Index] = FailedRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, exception.Message); + } } - // 2. Snapshot lifecycle. Reserve our own item handle, advise, pump - // until we see a fresh OnDataChange (or the deadline elapses), - // then tear it down. - int itemHandle = 0; - bool advised = false; + if (waiting.Count > 0) + { + WaitForSnapshots(serverHandle, waiting, results, timeout, pumpStep); + } + + foreach (SnapshotRead duplicate in sharing) + { + BulkReadResult? ownerResult = results[duplicate.SharedWith!.Index]; + if (ownerResult is null) + { + continue; + } + + BulkReadResult fanned = ownerResult.Clone(); + fanned.TagAddress = duplicate.TagAddress; + results[duplicate.Index] = fanned; + } + } + + private void WaitForSnapshots( + int serverHandle, + List waiting, + BulkReadResult?[] results, + TimeSpan timeout, + Action pumpStep) + { + // Phase 3: one shared deadline. Pump until every advised handle has + // seen its first OnDataChange since its baseline, or the deadline + // elapses. + int[] itemHandles = new int[waiting.Count]; + ulong[] baselines = new ulong[waiting.Count]; + for (int index = 0; index < waiting.Count; index++) + { + itemHandles[index] = waiting[index].ItemHandle; + baselines[index] = waiting[index].Baseline; + } + + bool[] received = new bool[waiting.Count]; + MxAccessValueCache.CachedValue[] values = new MxAccessValueCache.CachedValue[waiting.Count]; try { - itemHandle = AddItem(serverHandle, tagAddress); - ulong baseline = valueCache.CurrentVersion(serverHandle, itemHandle); - Advise(serverHandle, itemHandle); - advised = true; - DateTime deadline = DateTime.UtcNow + timeout; - bool gotValue = valueCache.TryWaitForUpdate( - serverHandle, - itemHandle, - baseline, - deadline, - pumpStep, - out MxAccessValueCache.CachedValue snapshot); - - return gotValue - ? SucceededRead(serverHandle, tagAddress, itemHandle, wasCached: false, snapshot) - : FailedRead( - serverHandle, - tagAddress, - itemHandle, - wasCached: false, - $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange."); + valueCache.TryWaitForUpdates(serverHandle, itemHandles, baselines, deadline, pumpStep, received, values); } catch (Exception exception) { - return FailedRead(serverHandle, tagAddress, itemHandle, wasCached: false, exception.Message); - } - finally - { - // Snapshot teardown — best-effort. Errors here are noted on the - // diagnostic message of the original result (above) by appending - // a cleanup suffix; we never re-throw from finally. - if (advised) + // A pump failure ends the wait for every tag still outstanding; + // tags captured before it keep their values. + for (int index = 0; index < waiting.Count; index++) { - try { UnAdvise(serverHandle, itemHandle); } catch { /* swallow — best effort */ } + if (!received[index]) + { + SnapshotRead failed = waiting[index]; + results[failed.Index] = FailedRead(serverHandle, failed.TagAddress, failed.ItemHandle, wasCached: false, exception.Message); + } + } + } + + for (int index = 0; index < waiting.Count; index++) + { + SnapshotRead snapshot = waiting[index]; + if (results[snapshot.Index] is not null) + { + continue; } - if (itemHandle != 0) + results[snapshot.Index] = received[index] + ? SucceededRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, values[index]) + : FailedRead( + serverHandle, + snapshot.TagAddress, + snapshot.ItemHandle, + wasCached: false, + $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange."); + } + } + + private void TearDownSnapshots( + int serverHandle, + List snapshots, + Action pumpStep) + { + // Phase 4: snapshot teardown — best-effort, never re-throws. Each + // distinct handle is unadvised and removed exactly once, even when + // MXAccess handed the same id to several tags in the batch. + HashSet unadvised = new(); + HashSet removed = new(); + int calls = 0; + foreach (SnapshotRead snapshot in snapshots) + { + if (snapshot.Advised && unadvised.Add(snapshot.ItemHandle)) { - try { RemoveItem(serverHandle, itemHandle); } catch { /* swallow — best effort */ } + PumpEvery(ref calls, pumpStep); + try { UnAdvise(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ } } } + + foreach (SnapshotRead snapshot in snapshots) + { + if (snapshot.ItemHandle != 0 && removed.Add(snapshot.ItemHandle)) + { + PumpEvery(ref calls, pumpStep); + try { RemoveItem(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ } + } + } + } + + private static void PumpEvery(ref int calls, Action pumpStep) + { + if (calls > 0 && calls % ReadBulkPumpInterval == 0) + { + try { pumpStep(); } catch { /* swallow — teardown is best effort */ } + } + + calls++; } private bool TryGetCachedReadFor( @@ -1308,6 +1495,41 @@ public sealed class MxAccessSession : IDisposable } } + /// + /// Mutable per-tag state for one ReadBulk snapshot lifecycle. A plain + /// class with get/set properties (no records or init accessors) so it + /// compiles under the worker's net48 target. + /// + private sealed class SnapshotRead + { + public SnapshotRead(int index, string tagAddress) + { + Index = index; + TagAddress = tagAddress; + } + + /// Gets the position of the tag in the request. + public int Index { get; } + + /// Gets the requested tag address. + public string TagAddress { get; } + + /// Gets or sets the snapshot item handle, 0 until AddItem succeeds. + public int ItemHandle { get; set; } + + /// Gets or sets the value-cache version captured before Advise. + public ulong Baseline { get; set; } + + /// Gets or sets a value indicating whether Advise succeeded. + public bool Advised { get; set; } + + /// + /// Gets or sets the earlier snapshot in the same batch whose + /// subscription this one shares (same item handle), or null. + /// + public SnapshotRead? SharedWith { get; set; } + } + private void ThrowIfDisposed() { if (disposed) diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 086188f..c66baaa 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -42,7 +42,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession /// Default alarm poll cadence when the environment says nothing usable. internal static readonly TimeSpan DefaultAlarmPollInterval = TimeSpan.FromMilliseconds(500); - private readonly TimeSpan alarmPollInterval = ResolveAlarmPollInterval(); + /// + /// Consecutive alarm-poll failures that fault the session. Matches the + /// default of + /// (AlarmFailoverConfig.consecutive_failure_threshold, default 3): + /// one transient COM hiccup in PollOnce is logged and retried on + /// the next tick, and only a run of failures with no success in between + /// is treated as a dead subscription. + /// + internal const int DefaultAlarmPollFailureThreshold = 3; private readonly IMxAccessComObjectFactory factory; private readonly IMxAccessEventSink eventSink; @@ -221,6 +229,28 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession /// internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout(); + /// + /// Cadence the alarm poll loop waits between polls. Resolved from the + /// launcher environment; internal-settable as a test seam (set before + /// ). + /// + internal TimeSpan AlarmPollInterval { get; set; } = ResolveAlarmPollInterval(); + + /// + /// Consecutive alarm-poll failures that fault the session; clamped to a + /// minimum of 1. Internal-settable as a test seam. + /// + internal int AlarmPollFailureThreshold { get; set; } = DefaultAlarmPollFailureThreshold; + + /// + /// Sink for the transient alarm-poll failure warning. Defaults to + /// , the stream WorkerConsoleLogger + /// writes to. The line carries the exception type, HRESULT, and + /// failure counts only — never the exception message, which can echo + /// alarm or tag names. Tests substitute a collector. + /// + internal Action? AlarmPollWarningSink { get; set; } + /// /// Resolves the write-completion wait from the launcher-provided /// environment variable; a missing or invalid value falls back to @@ -288,6 +318,13 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession { staRuntime.Start(); + // WRK-02: a post-start STA loop death has no other observer. Record it + // on the event queue so the IPC drain loop sends a WorkerFault to the + // gateway and ends the pipe session (and with it the worker process), + // instead of leaving a live process whose commands queue forever. + staRuntime.RegisterLoopFaultHandler( + exception => eventQueue.RecordFault(CreateStaLoopFault(exception))); + WorkerReady ready = await staRuntime.InvokeAsync( () => { @@ -345,13 +382,16 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession IAlarmCommandHandler handler, CancellationToken cancellationToken) { + TimeSpan pollInterval = AlarmPollInterval; + int failureThreshold = AlarmPollFailureThreshold < 1 ? 1 : AlarmPollFailureThreshold; return Task.Run(async () => { + int consecutiveFailures = 0; while (!cancellationToken.IsCancellationRequested) { try { - await Task.Delay(alarmPollInterval, cancellationToken).ConfigureAwait(false); + await Task.Delay(pollInterval, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -384,6 +424,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession } }, cancellationToken).ConfigureAwait(false); + consecutiveFailures = 0; } catch (OperationCanceledException) { @@ -413,19 +454,66 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession // A real alarm-poll failure (COMException from // GetXmlCurrentAlarms2, malformed-XML parse failure, an // STA-affinity InvalidOperationException from - // EnsureOnAlarmConsumerThread, etc.). Record it as a - // fault on the event queue so a broken alarm subscription - // — or an affinity-invariant regression — becomes + // EnsureOnAlarmConsumerThread, etc.). A single failure is + // logged and retried on the next tick: one transient COM + // hiccup must not tear down the whole session (every data + // subscription with it). Only a run of + // AlarmPollFailureThreshold failures with no success in + // between is recorded as a fault on the event queue, so a + // broken alarm subscription — or an affinity-invariant + // regression, which fails every tick — still becomes // observable on the IPC fault path instead of silently // faulting this never-awaited task. The loop then stops — // the subscription is dead. - eventQueue.RecordFault(CreateAlarmPollFault(exception)); - return; + if (staRuntime.LoopFault is not null) + { + // The STA loop itself died; its handler already recorded + // the StaHung fault. This failure is just the fast-fail + // InvokeAsync returns afterwards — recording it would add + // a second, mislabelled alarm-poll fault. + return; + } + + consecutiveFailures++; + if (consecutiveFailures >= failureThreshold) + { + eventQueue.RecordFault(CreateAlarmPollFault(exception)); + return; + } + + WriteAlarmPollWarning(exception, consecutiveFailures, failureThreshold); } } }, CancellationToken.None); } + private void WriteAlarmPollWarning(Exception exception, int consecutiveFailures, int failureThreshold) + { + // Same "level= event= k=v" line shape as WorkerConsoleLogger and + // the wnwrap truncation warning. Type and HRESULT only — the exception + // message is deliberately left out (it can carry alarm or tag names). + string hresult = exception is System.Runtime.InteropServices.COMException comException + ? "0x" + comException.HResult.ToString("X8", System.Globalization.CultureInfo.InvariantCulture) + : "none"; + string message = string.Format( + System.Globalization.CultureInfo.InvariantCulture, + "level=Warning event=AlarmPollFailed exceptionType={0} hresult={1} consecutiveFailures={2} faultThreshold={3}", + exception.GetType().FullName, + hresult, + consecutiveFailures, + failureThreshold); + + try + { + Action sink = AlarmPollWarningSink ?? Console.Error.WriteLine; + sink(message); + } + catch + { + // Logging must never end the poll loop. + } + } + private void EnsureOnAlarmConsumerThread() { AssertOnAlarmConsumerThread(alarmConsumerThreadId, Environment.CurrentManagedThreadId); @@ -458,6 +546,41 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession } } + /// + /// Builds the fault reported when the STA loop dies after start. Uses + /// : the gateway already treats + /// that category as "this worker's apartment can no longer run COM + /// work", which is exactly the state a dead loop leaves behind. + /// + internal static WorkerFault CreateStaLoopFault(Exception exception) + { + // Type and HRESULT only, never exception.Message — the same type-only + // rule every other worker failure line follows. + string message = string.Format( + System.Globalization.CultureInfo.InvariantCulture, + "Worker STA loop terminated unexpectedly: exceptionType={0} hresult=0x{1:X8}", + exception.GetType().FullName, + exception.HResult); + WorkerFault fault = new() + { + Category = WorkerFaultCategory.StaHung, + ExceptionType = exception.GetType().FullName ?? string.Empty, + DiagnosticMessage = message, + ProtocolStatus = new ProtocolStatus + { + Code = ProtocolStatusCode.WorkerUnavailable, + Message = message, + }, + }; + + if (exception is System.Runtime.InteropServices.COMException comException) + { + fault.Hresult = comException.HResult; + } + + return fault; + } + private static WorkerFault CreateAlarmPollFault(Exception exception) { string message = diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs index 3841d5f..652b7eb 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs @@ -162,6 +162,46 @@ public sealed class MxAccessValueCache } } + /// + /// Removes every cache slot recorded under a server handle. The session + /// calls this when a server handle is unregistered: MXAccess drops all of + /// that server's items with it, so the per-item + /// never runs for them and their entries would otherwise outlive the + /// handle (and could be served to a later server that reuses the id). + /// + /// MXAccess server handle that was unregistered. + public void RemoveServer(int serverHandle) + { + lock (syncRoot) + { + List stale = new(); + foreach (long key in entries.Keys) + { + if ((int)(key >> 32) == serverHandle) + { + stale.Add(key); + } + } + + foreach (long key in stale) + { + entries.Remove(key); + } + } + } + + /// Gets the number of handle pairs with a cached value (test seam). + internal int Count + { + get + { + lock (syncRoot) + { + return entries.Count; + } + } + } + /// /// Waits until the cache entry's version exceeds /// or the deadline elapses, calling on every @@ -231,6 +271,117 @@ public sealed class MxAccessValueCache } } + /// + /// Multi-handle form of : waits until every + /// handle in has an entry newer than its + /// matching slot, or the single shared + /// deadline elapses. Each handle is captured the first time an update is + /// observed for it (its first OnDataChange since the baseline), exactly + /// as a dedicated per-handle wait would have captured it, so a bulk + /// snapshot read waits at most one timeout in total rather than one per + /// tag. + /// + /// + /// Pumps unconditionally at the top of every iteration and blocks the + /// idle part of each iteration in + /// — the same pump + /// discipline as , for the same reason. + /// + /// MXAccess server handle shared by every item. + /// Item handles to wait on. + /// Per-handle version baselines captured before advising. + /// Absolute UTC deadline shared by every handle. + /// Action that pumps any pending Windows messages. + /// Output: at index i when handle i received an update before the deadline. + /// Output: the captured value at index i when [i] is set. + /// Upper bound on one idle wait slice. + /// The number of handles that received an update before the deadline. + public int TryWaitForUpdates( + int serverHandle, + IReadOnlyList itemHandles, + IReadOnlyList sinceVersions, + DateTime deadlineUtc, + Action pumpStep, + bool[] received, + CachedValue[] values, + int pollIntervalMs = 5) + { + if (itemHandles is null) + { + throw new ArgumentNullException(nameof(itemHandles)); + } + + if (sinceVersions is null) + { + throw new ArgumentNullException(nameof(sinceVersions)); + } + + if (pumpStep is null) + { + throw new ArgumentNullException(nameof(pumpStep)); + } + + if (received is null) + { + throw new ArgumentNullException(nameof(received)); + } + + if (values is null) + { + throw new ArgumentNullException(nameof(values)); + } + + int count = itemHandles.Count; + if (sinceVersions.Count != count || received.Length < count || values.Length < count) + { + throw new ArgumentException("Handle, baseline, and output arrays must have matching lengths."); + } + + int receivedCount = 0; + for (int index = 0; index < count; index++) + { + if (received[index]) + { + receivedCount++; + } + } + + while (true) + { + pumpStep(); + + for (int index = 0; index < count; index++) + { + if (received[index]) + { + continue; + } + + if (TryGet(serverHandle, itemHandles[index], out CachedValue value) && value.Version > sinceVersions[index]) + { + received[index] = true; + values[index] = value; + receivedCount++; + } + } + + if (receivedCount >= count) + { + return receivedCount; + } + + TimeSpan remaining = deadlineUtc - DateTime.UtcNow; + if (remaining <= TimeSpan.Zero) + { + return receivedCount; + } + + StaWaitHelper.WaitForSignalOrMessages( + updateSignal, + StaWaitHelper.ClampWaitMilliseconds(remaining, pollIntervalMs)); + } + } + /// Returns the current version for a handle pair, or 0 if no entry exists. /// MXAccess server handle. /// MXAccess item handle. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/IStaMessagePump.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/IStaMessagePump.cs new file mode 100644 index 0000000..c850d12 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/IStaMessagePump.cs @@ -0,0 +1,22 @@ +using System; +using System.Threading; + +namespace ZB.MOM.WW.MxGateway.Worker.Sta; + +/// +/// The two Win32 pump operations drives on its +/// STA thread. is the only production +/// implementation; the seam exists so Worker.Tests can observe pump +/// interleaving and inject a pump failure without a real message queue. +/// +internal interface IStaMessagePump +{ + /// Waits for a command wake event or Windows messages. + /// Event signalled when command work is available. + /// Maximum time to wait. + void WaitForWorkOrMessages(WaitHandle commandWakeEvent, TimeSpan timeout); + + /// Pumps and dispatches all pending Windows messages. + /// The number of messages dispatched. + int PumpPendingMessages(); +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs index 535eec3..8e6e758 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs @@ -6,7 +6,7 @@ using Microsoft.Win32.SafeHandles; namespace ZB.MOM.WW.MxGateway.Worker.Sta; /// Pumps Windows messages on the STA thread to allow MXAccess COM events to deliver. -public sealed class StaMessagePump +public sealed class StaMessagePump : IStaMessagePump { private const uint Infinite = 0xFFFFFFFF; private const uint PmRemove = 0x0001; diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs index f8a83c1..ac16041 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs @@ -8,7 +8,7 @@ namespace ZB.MOM.WW.MxGateway.Worker.Sta; public sealed class StaRuntime : IDisposable { private readonly IStaComApartmentInitializer comApartmentInitializer; - private readonly StaMessagePump messagePump; + private readonly IStaMessagePump messagePump; private readonly ConcurrentQueue commandQueue = new(); private readonly AutoResetEvent commandWakeEvent = new(false); private readonly ManualResetEventSlim startedEvent = new(false); @@ -20,6 +20,8 @@ public sealed class StaRuntime : IDisposable private bool startRequested; private bool shutdownRequested; private Exception? startupException; + private Exception? loopFault; + private Action? loopFaultHandler; private long lastActivityUtcTicks; private bool comInitialized; @@ -41,6 +43,21 @@ public sealed class StaRuntime : IDisposable IStaComApartmentInitializer comApartmentInitializer, StaMessagePump messagePump, TimeSpan idlePumpInterval) + : this(comApartmentInitializer, (IStaMessagePump)messagePump, idlePumpInterval) + { + } + + /// + /// Initializes a new instance of over any + /// (test seam). + /// + /// COM apartment initializer. + /// Message pump for the STA thread. + /// Interval for idle message pump waits. + internal StaRuntime( + IStaComApartmentInitializer comApartmentInitializer, + IStaMessagePump messagePump, + TimeSpan idlePumpInterval) { this.comApartmentInitializer = comApartmentInitializer ?? throw new ArgumentNullException(nameof(comApartmentInitializer)); @@ -79,6 +96,59 @@ public sealed class StaRuntime : IDisposable /// public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet; + /// + /// Gets the exception that terminated the STA loop after a successful + /// , or while the loop is healthy + /// (or stopped by a requested shutdown). Startup failures are not reported + /// here — throws them. + /// + public Exception? LoopFault + { + get + { + lock (gate) + { + return loopFault; + } + } + } + + /// + /// Registers the callback invoked once, on the dying STA thread, when the + /// STA loop terminates with an exception after a successful + /// . When the loop has already died the callback runs + /// immediately on the calling thread, so a fault that lands between + /// and registration is never lost. Only one handler is + /// kept; a later registration replaces an earlier one. + /// + /// + /// Without this, a post-start loop death (a failed + /// MsgWaitForMultipleObjectsEx, or an exception escaping a dispatched + /// window message) left the worker process alive with no STA: every later + /// command queued forever and the gateway only learned of it through the + /// heartbeat watchdog, if at all. + /// + /// Callback receiving the terminating exception. + public void RegisterLoopFaultHandler(Action handler) + { + if (handler is null) + { + throw new ArgumentNullException(nameof(handler)); + } + + Exception? alreadyFaulted; + lock (gate) + { + loopFaultHandler = handler; + alreadyFaulted = shutdownRequested ? null : loopFault; + } + + if (alreadyFaulted is not null) + { + handler(alreadyFaulted); + } + } + /// /// Pumps any pending Windows messages on the calling thread and refreshes /// the STA activity timestamp. Intended for commands that synchronously @@ -176,6 +246,14 @@ public sealed class StaRuntime : IDisposable lock (gate) { + if (loopFault is not null) + { + // The loop is gone; nothing would ever dequeue this item. + return Task.FromException(new InvalidOperationException( + "The worker STA loop terminated unexpectedly.", + loopFault)); + } + if (shutdownRequested) { return Task.FromException(new StaRuntimeShutdownException()); @@ -244,12 +322,14 @@ public sealed class StaRuntime : IDisposable private void ThreadMain() { + bool started = false; try { StaThreadId = Thread.CurrentThread.ManagedThreadId; comApartmentInitializer.Initialize(); comInitialized = true; MarkActivity(); + started = true; startedEvent.Set(); while (!IsShutdownRequested()) @@ -264,8 +344,44 @@ public sealed class StaRuntime : IDisposable } catch (Exception exception) { - startupException = exception; - startedEvent.Set(); + if (!started) + { + // Start() is blocked on startedEvent and rethrows this. + startupException = exception; + startedEvent.Set(); + } + else + { + // Post-start loop death. Start() returned long ago, so nothing + // would ever read startupException: record the fault where + // InvokeAsync and the registered handler see it. Set under the + // gate so every item enqueued before it is still in the queue + // for the finally block's CancelQueuedCommands, and every + // InvokeAsync after it fails fast instead of queueing forever. + Action? handler; + lock (gate) + { + loopFault = exception; + + // A loop that throws while a requested shutdown is already + // in progress is being torn down on purpose; reporting it + // as a worker fault would turn a clean stop into a fault. + handler = shutdownRequested ? null : loopFaultHandler; + } + + if (handler is not null) + { + try + { + handler(exception); + } + catch + { + // The handler reports the fault; a failure inside it + // must not skip the apartment cleanup below. + } + } + } } finally { @@ -292,6 +408,17 @@ public sealed class StaRuntime : IDisposable MarkActivity(); workItem.Execute(); MarkActivity(); + + // Pump between work items, not only once the whole queue is + // drained: a burst of queued commands (bulk callers, the alarm + // poll, a reconnecting client replaying subscriptions) would + // otherwise hold every inbound COM callback — OnDataChange, + // OnWriteComplete — behind the entire backlog. PeekMessage with an + // empty queue is cheap, so the per-item cost is negligible. + if (!commandQueue.IsEmpty) + { + messagePump.PumpPendingMessages(); + } } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs b/src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs index cdf6e48..2642607 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs @@ -14,10 +14,61 @@ public static class WorkerApplication /// The process exit code. public static int Run(string[] args) { + WorkerConsoleLogger logger = new(Console.Error); + + // WRK-09: installed only on the production entry point (not the + // injectable overloads tests call) because the hook is process-global. + AppDomain.CurrentDomain.UnhandledException += (_, eventArgs) => + HandleUnhandledException( + logger, + eventArgs.ExceptionObject, + eventArgs.IsTerminating, + Environment.Exit); + return Run( args, new EnvironmentVariableWorkerEnvironment(), - new WorkerConsoleLogger(Console.Error)); + logger); + } + + /// + /// Body of the worker's AppDomain.CurrentDomain.UnhandledException + /// hook. Logs one WorkerUnhandledException line and exits the + /// process with , so the + /// gateway's launcher sees a defined worker exit code rather than the + /// CLR's crash code. + /// + /// + /// Logs the exception type only — never its message or stack, which + /// can carry tag values or credentials passed through a command — the + /// same rule every other worker failure line follows. A logging + /// failure is swallowed so the exit always happens. + /// + /// Worker logger. + /// The unhandled exception object (not necessarily an ). + /// Whether the CLR is terminating the process. + /// Process-exit action; in production. + internal static void HandleUnhandledException( + IWorkerLogger logger, + object? exceptionObject, + bool isTerminating, + Action exit) + { + try + { + logger.Error("WorkerUnhandledException", new Dictionary + { + ["exit_code"] = WorkerExitCode.UnhandledException, + ["exception_type"] = exceptionObject?.GetType().FullName ?? "null", + ["is_terminating"] = isTerminating, + }); + } + catch + { + // Never let a logging failure skip the defined exit. + } + + exit((int)WorkerExitCode.UnhandledException); } /// Initializes and runs the worker with custom environment and logging.