Compare commits

..

7 Commits

Author SHA1 Message Date
Joseph Doherty 6f4c7c3bf1 Keep subtag acks and alarm reads safe on the managed ack path
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (pull_request) Successful in 2m0s
ci / nightly-windev (pull_request) Has been skipped
ci / java (pull_request) Successful in 2m16s
ci / java (push) Successful in 2m19s
ci / portable (push) Failing after 2m31s
ci / portable (pull_request) Failing after 2m36s
ci / windows-x86 (push) Successful in 3m11s
Review fixes for the opt-in managed GUID ack:

- Subtag-mode GUID acks (forced subtag, or failover on the standby) stay on
  the consumer; their GUIDs are synthetic and only the subtag consumer can
  resolve them. Forced-subtag sessions no longer build a managed client.
- A failed managed Initialize/Register/Subscribe drops back to the wnwrap
  ack path instead of failing the alarm subscription.
- Drop stale crefs to a nonexistent Unverified member in
  WnalNativeAckConsumer.
- Unit tests for routing, fallback, and disposal; document
  MXGATEWAY_USE_MANAGED_ALARM_ACK.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 13:11:01 -04:00
Joseph Doherty 6affbd57ea feat(worker): add opt-in managed GUID alarm-ack path
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m19s
ci / java (push) Successful in 2m8s
ci / portable (push) Failing after 2m23s
ci / nightly-windev (pull_request) Has been skipped
ci / windows-x86 (pull_request) Successful in 1m18s
ci / java (pull_request) Successful in 2m3s
ci / portable (pull_request) Failing after 2m17s
Adds a managed acknowledge path over AVEVA's modern alarm client so GUID
acks carry full operator identity (domain + full name) and work on builds
where the wnwrap COM AlarmAckByGUID is E_NOTIMPL. Default off; opt in with
MXGATEWAY_USE_MANAGED_ALARM_ACK=1 (wnwrap behavior unchanged otherwise).

- AaManagedAlarmAckConsumer: reflection wrapper over
  aaAlarmManagedClient.AlarmClient (InitializeConsumer/RegisterConsumer/
  Subscribe/AlarmAckByGUID); EnvGatedFactory degrades to null (wnwrap) if the
  DLL can't load, so alarm reads never break.
- IManagedAlarmAckClient seam. AlarmDispatcher takes an optional client and
  routes GUID acks to it (subscribing + disposing it alongside the reader).
- AlarmCommandHandler gains an optional managedAckFactory; WorkerPipeSession
  wires AaManagedAlarmAckConsumer.EnvGatedFactory at the production sites.
- WnalNativeAckConsumer: confirmed native alarmlst.dll WNAL submit layer
  (WNALalarmReqSetText / WNALUserRequest / WNALUserRequestEx + lpAlmAck
  layout). List acquisition is intentionally not hand-rolled: the alarm-record
  layout and selection semantics are provider-defined and not recoverable from
  alarmlst.dll, so inferring them risks acking the wrong alarm.
- AlarmManagedGuidAckLiveSmokeTests: skip-gated live smoke that pulls a real
  (non-degraded) alarm GUID and asserts the ack APPLIES (Acknowledge
  transition / acked snapshot), not merely rc=0.

By-name acks stay on wnwrap (AlarmClient has no AlarmAckByName). Grounded in
static RE of alarmlst.dll, aaAlarmManagedClient.dll and alarmmgr.exe.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-10-18 12:42:07 -04:00
dohertj2 6fc423f608 Merge pull request 'Retain fake worker fault diagnostics in test fixtures' (#155) from fix/fake-worker-fault-diagnostics into main
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m17s
ci / java (push) Successful in 2m3s
ci / portable (push) Successful in 8m23s
2026-10-13 10:46:14 -04:00
Joseph Doherty 83aa76c33d Retain fake worker fault diagnostics in test fixtures
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m20s
ci / java (push) Successful in 2m19s
ci / windows-x86 (pull_request) Successful in 1m20s
ci / nightly-windev (pull_request) Has been skipped
ci / java (pull_request) Successful in 2m23s
ci / portable (push) Successful in 9m10s
ci / portable (pull_request) Successful in 9m2s
2026-10-13 10:35:28 -04:00
dohertj2 d779daf6af Merge pull request 'Make framed worker and readiness test gates deterministic' (#153) from fix/framed-worker-test-gates into main
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m22s
ci / java (push) Successful in 2m2s
ci / portable (push) Successful in 8m26s
2026-10-13 09:59:45 -04:00
Joseph Doherty 42c120aa83 Preserve framed reads across fake worker emission gates
ci / windows-x86 (push) Successful in 1m20s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m17s
ci / windows-x86 (pull_request) Successful in 1m25s
ci / nightly-windev (pull_request) Has been skipped
ci / java (pull_request) Successful in 2m27s
ci / portable (push) Successful in 9m43s
ci / portable (pull_request) Successful in 9m47s
2026-10-13 09:42:09 -04:00
dohertj2 19920b13da Merge pull request 'Allow redundant must_use only in generated Rust service traits' (#152) from fix/generated-rust-must-use into main
ci / windows-x86 (push) Successful in 1m25s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m15s
ci / portable (push) Successful in 9m6s
2026-10-13 09:37:53 -04:00
17 changed files with 1573 additions and 76 deletions
+22
View File
@@ -793,6 +793,28 @@ in the codebase for the forward-compat shape, but the gateway-side
`AcknowledgeAlarmByName` when the public RPC supplies a recognizable
`Provider!Group.Tag` reference.
**Opt-in managed GUID ack.** Setting `MXGATEWAY_USE_MANAGED_ALARM_ACK=1`
in the gateway's environment (workers inherit it) routes GUID-keyed acks
through `aaAlarmManagedClient.AlarmClient.AlarmAckByGUID` instead of the
wnwrap method. That managed method is implemented on this build and
carries operator domain and full name natively. The worker reaches it by
reflection (`AaManagedAlarmAckConsumer`) because the members are
non-public, and runs its own Initialize → Register → Subscribe lifecycle
against the alarm subscription expression. The path is deliberately
conservative:
- If the DLL cannot load or the lifecycle returns a non-zero status, the
worker drops back to the wnwrap ack path; alarm reads never fail
because of it.
- Subtag-mode acks (forced subtag, or failover currently on the standby)
stay on the subtag consumer: their GUIDs are synthetic and only it can
resolve them.
- By-name acks stay on wnwrap — `AlarmClient` has no by-name ack.
A zero return still means *accepted*, not *applied*; a secured
(`MxSecurityOperate`) alarm needs an authenticated operator. The skip-gated
`AlarmManagedGuidAckLiveSmokeTests` asserts the ack actually lands.
### 5. STA / threading — resolved
The wnwrap COM is `ThreadingModel=Apartment`. The consumer's
+7
View File
@@ -446,6 +446,13 @@ behavior.
The alarm monitor is independent of client sessions: `AcknowledgeAlarm` and
`StreamAlarms` are session-less RPCs served by the monitor.
GUID-keyed acknowledges can opt into AVEVA's managed alarm client by setting
the environment variable `MXGATEWAY_USE_MANAGED_ALARM_ACK=1` on the gateway
process (workers inherit it; under NSSM, add it to the service environment).
It is off by default and has no `MxGateway:*` option because it is still
awaiting live confirmation on the rig. See
[AlarmClientDiscovery](./AlarmClientDiscovery.md) ("Opt-in managed GUID ack").
### Alarm fallback options
The `Fallback` sub-section controls how the alarm feed selects between the
@@ -2,6 +2,7 @@ using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Xunit.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
@@ -21,7 +22,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
/// the fake worker harness. Covers fan-out to two concurrent subscribers, independent
/// cancellation isolation, and the per-session subscriber cap.
/// </summary>
public sealed class GatewayEndToEndMultiSubscriberTests
public sealed class GatewayEndToEndMultiSubscriberTests(ITestOutputHelper output)
{
private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
private const int ServerHandle = 3001;
@@ -37,7 +38,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
public async Task StreamEvents_TwoSubscribers_BothReceiveAllEvents()
{
MultiEventFakeWorkerProcessLauncher launcher = new(EventCount);
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher);
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine);
OpenSessionReply openReply = await fixture.Service.OpenSession(
new OpenSessionRequest
@@ -136,7 +137,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
public async Task StreamEvents_OneSubscriberCancels_OtherContinuesReceivingEvents()
{
GatedEventFakeWorkerProcessLauncher launcher = new();
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher);
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine);
OpenSessionReply openReply = await fixture.Service.OpenSession(
new OpenSessionRequest
@@ -254,7 +255,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
// (EventSubscriberLimitReached) is only reachable when multi-subscriber mode is on.
// Without it the service hits EventSubscriberAlreadyActive instead — also
// ResourceExhausted, but for the wrong reason — masking a misconfiguration.
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap);
await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap, diagnosticOutput: output.WriteLine);
OpenSessionReply openReply = await fixture.Service.OpenSession(
new OpenSessionRequest
@@ -413,20 +414,25 @@ public sealed class GatewayEndToEndMultiSubscriberTests
{
private readonly GatewayMetrics _metrics = new();
private readonly SessionRegistry _registry = new();
private readonly FakeWorkerDiagnostics _diagnostics;
/// <summary>Initializes a new instance of the <see cref="MultiSubscriberGatewayServiceFixture"/> class.</summary>
/// <param name="launcher">Fake worker process launcher backing the session manager.</param>
/// <param name="diagnosticOutput">Safe metadata output sink.</param>
/// <param name="maxEventSubscribersPerSession">Maximum concurrent event subscribers allowed per session.</param>
public MultiSubscriberGatewayServiceFixture(
IWorkerProcessLauncher launcher,
int maxEventSubscribersPerSession = 8)
int maxEventSubscribersPerSession = 8,
Action<string>? diagnosticOutput = null)
{
_diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics;
_diagnostics.SetOutput(diagnosticOutput);
IOptions<GatewayOptions> options = Options.Create(CreateOptions(maxEventSubscribersPerSession));
SessionWorkerClientFactory workerClientFactory = new(
launcher,
options,
_metrics,
NullLoggerFactory.Instance);
_diagnostics);
SessionManager sessionManager = new(
_registry,
workerClientFactory,
@@ -499,13 +505,20 @@ public sealed class GatewayEndToEndMultiSubscriberTests
/// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
// Session disposal and worker observation share the existing test budget.
// The observer consumes a late cleanup fault without replacing the primary failure.
_diagnostics.Track(DisposeSessionsAsync(), "teardown");
await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false);
_metrics.Dispose();
}
private async Task DisposeSessionsAsync()
{
foreach (GatewaySession session in _registry.Snapshot())
{
await session.DisposeAsync();
await session.DisposeAsync().ConfigureAwait(false);
}
_metrics.Dispose();
}
private static GatewayOptions CreateOptions(int maxEventSubscribersPerSession) =>
@@ -540,7 +553,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
/// command, then waits for shutdown. Events carry an indexed string value so the test can
/// verify fan-out order across two subscribers.
/// </summary>
private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher
private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher, IDiagnosticFakeWorker
{
public const int ProcessId = 7710;
@@ -549,12 +562,16 @@ public sealed class GatewayEndToEndMultiSubscriberTests
/// <summary>Gets the task representing the fake worker's running background loop.</summary>
public Task WorkerTask { get; private set; } = Task.CompletedTask;
/// <inheritdoc />
public FakeWorkerDiagnostics Diagnostics { get; } = new();
/// <inheritdoc />
public Task<WorkerProcessHandle> LaunchAsync(
WorkerProcessLaunchRequest request,
CancellationToken cancellationToken = default)
{
WorkerTask = RunWorkerAsync(request, cancellationToken);
Diagnostics.Track(WorkerTask);
return Task.FromResult(new WorkerProcessHandle(
_process,
@@ -578,6 +595,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
{
WorkerEnvelope envelope = await harness.ReadGatewayEnvelopeAsync(cancellationToken).ConfigureAwait(false);
Diagnostics.Received(envelope);
if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
{
await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -630,7 +648,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
/// deterministic. Call <see cref="StopEmitting"/> before closing the session so the
/// worker loop exits cleanly and can process the shutdown envelope.
/// </summary>
private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher, IDiagnosticFakeWorker
{
public const int ProcessId = 7720;
@@ -643,6 +661,9 @@ public sealed class GatewayEndToEndMultiSubscriberTests
/// <summary>Gets the task representing the fake worker's running background loop.</summary>
public Task WorkerTask { get; private set; } = Task.CompletedTask;
/// <inheritdoc />
public FakeWorkerDiagnostics Diagnostics { get; } = new();
/// <summary>Releases the gate so the worker emits one event.</summary>
public void AllowNextEvent() => _emitGate.Release();
@@ -662,6 +683,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
CancellationToken cancellationToken = default)
{
WorkerTask = RunWorkerAsync(request, cancellationToken);
Diagnostics.Track(WorkerTask);
return Task.FromResult(new WorkerProcessHandle(
_process,
@@ -685,19 +707,19 @@ public sealed class GatewayEndToEndMultiSubscriberTests
int advisedItemHandle = 0;
int emittedCount = 0;
// Read envelopes one at a time. Between each envelope, if we have a
// subscription and the gate is ready, emit an event before reading the
// next envelope. When _stopEmitting is set, drain the gate and read
// remaining envelopes (including shutdown) without emitting.
await using GatedWorkerEnvelopeReader incoming = new(
harness.ReadGatewayEnvelopeAsync, _emitGate, cancellationToken);
while (!cancellationToken.IsCancellationRequested)
{
// While subscribed and not stopped, try to emit gated events using
// a short-timeout peek at the gate — yield to incoming envelopes to
// avoid starving shutdown processing.
while (advisedServerHandle != 0
&& !_stopEmitting
&& await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
WorkerEnvelope? envelope = await incoming.ReadOrGateAsync(
advisedServerHandle != 0 && !_stopEmitting).ConfigureAwait(false);
if (envelope is null)
{
if (_stopEmitting)
{
continue;
}
int index = ++emittedCount;
await harness.EmitEventAsync(
MxEventFamily.OnDataChange,
@@ -714,23 +736,10 @@ public sealed class GatewayEndToEndMultiSubscriberTests
};
mxEvent.OnDataChange = new OnDataChangeEvent();
}).ConfigureAwait(false);
}
// Use a short timeout so the emit loop above is re-evaluated
// periodically — but long enough not to spam.
WorkerEnvelope? envelope = null;
try
{
using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Timed out waiting for an envelope — loop back to check gate / emit.
continue;
}
Diagnostics.Received(envelope);
if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
{
await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -1,8 +1,10 @@
using System.Collections.Concurrent;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Microsoft.Extensions.Time.Testing;
using Xunit.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
@@ -36,7 +38,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
/// per session and survives detach, so events emitted while no subscriber is attached are
/// retained for the reconnecting stream.
/// </remarks>
public sealed class GatewayEndToEndReconnectReplayTests
public sealed class GatewayEndToEndReconnectReplayTests(ITestOutputHelper output)
{
private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
private const int ServerHandle = 3001;
@@ -58,7 +60,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
GatedEventFakeWorkerProcessLauncher launcher = new();
// Capacity 16 retains every event emitted here (7 total), so nothing is evicted and a
// resume from a middle cursor never produces a gap.
await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16);
await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16, diagnosticOutput: output.WriteLine);
string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap");
@@ -157,7 +159,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
GatedEventFakeWorkerProcessLauncher launcher = new();
// Small capacity forces eviction: with 6 events emitted and a 3-slot ring, the first 3
// are evicted and only the newest 3 remain replayable.
await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity);
await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity, diagnosticOutput: output.WriteLine);
string sessionId = await OpenSessionAsync(fixture, "reconnect-gap");
@@ -256,7 +258,8 @@ public sealed class GatewayEndToEndReconnectReplayTests
launcher,
replayBufferCapacity: 16,
replayRetentionSeconds: retentionSeconds,
timeProvider: clock);
timeProvider: clock,
diagnosticOutput: output.WriteLine);
string sessionId = await OpenSessionAsync(fixture, "reconnect-aged-out");
@@ -340,6 +343,34 @@ public sealed class GatewayEndToEndReconnectReplayTests
$"Live event {delivered.WorkerSequence} must be newer than the highest previously seen {highestSeen}.");
}
/// <summary>Setup failures preserve gateway EOF and report the original fake worker fault safely.</summary>
/// <param name="faultCommand">The setup command where the fake worker fails.</param>
/// <returns>The asynchronous test operation.</returns>
[Theory]
[InlineData(MxCommandKind.Register)]
[InlineData(MxCommandKind.AddItem)]
public async Task Setup_WhenFakeWorkerFaults_ReportsFaultWithoutMaskingRpcFailure(MxCommandKind faultCommand)
{
ConcurrentQueue<string> output = new();
GatedEventFakeWorkerProcessLauncher launcher = new(faultCommand);
Exception? failure = await Record.ExceptionAsync(async () =>
{
await using ReconnectReplayGatewayServiceFixture fixture = new(
launcher, replayBufferCapacity: 16, diagnosticOutput: output.Enqueue);
string sessionId = await OpenSessionAsync(fixture, "injected-fault");
await WireUpAdviseAsync(fixture, sessionId);
});
RpcException rpc = Assert.IsType<RpcException>(failure);
Assert.Equal(StatusCode.Unavailable, rpc.StatusCode);
Assert.Contains("Worker pipe disconnected", rpc.Status.Detail);
Assert.Contains(output, line => line.Contains("source=fake-worker")
&& line.Contains("type=WorkerFrameProtocolException") && line.Contains("code=InvalidEnvelope")
&& line.Contains($"command={faultCommand}") && line.Contains("phase=command"));
Assert.Contains(output, line => line.Contains("source=gateway") && line.Contains("code=EndOfStream"));
Assert.DoesNotContain(output, line => line.Contains("secret-item") || line.Contains("Galaxy.Tag.Value"));
}
// ---- shared flow helpers ----
private static async Task<string> OpenSessionAsync(
@@ -490,6 +521,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
{
private readonly GatewayMetrics _metrics = new();
private readonly SessionRegistry _registry = new();
private readonly FakeWorkerDiagnostics _diagnostics;
/// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
/// <param name="launcher">Fake worker process launcher backing the session manager.</param>
@@ -499,6 +531,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
/// duration of a fast test; the age-eviction test shortens it and drives
/// <paramref name="timeProvider"/>.
/// </param>
/// <param name="diagnosticOutput">Safe metadata output sink.</param>
/// <param name="timeProvider">
/// Clock handed to the session manager, which flows through to the session's event
/// distributor. Pass a fake to make age-eviction deterministic.
@@ -507,15 +540,18 @@ public sealed class GatewayEndToEndReconnectReplayTests
IWorkerProcessLauncher launcher,
int replayBufferCapacity,
double replayRetentionSeconds = 300,
TimeProvider? timeProvider = null)
TimeProvider? timeProvider = null,
Action<string>? diagnosticOutput = null)
{
_diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics;
_diagnostics.SetOutput(diagnosticOutput);
IOptions<GatewayOptions> options = Options.Create(
CreateOptions(replayBufferCapacity, replayRetentionSeconds));
SessionWorkerClientFactory workerClientFactory = new(
launcher,
options,
_metrics,
NullLoggerFactory.Instance);
_diagnostics);
SessionManager sessionManager = new(
_registry,
workerClientFactory,
@@ -585,13 +621,20 @@ public sealed class GatewayEndToEndReconnectReplayTests
/// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
// Session disposal and worker observation share the existing test budget.
// The observer consumes a late cleanup fault without replacing the primary failure.
_diagnostics.Track(DisposeSessionsAsync(), "teardown");
await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false);
_metrics.Dispose();
}
private async Task DisposeSessionsAsync()
{
foreach (GatewaySession session in _registry.Snapshot())
{
await session.DisposeAsync().ConfigureAwait(false);
}
_metrics.Dispose();
}
private static GatewayOptions CreateOptions(int replayBufferCapacity, double replayRetentionSeconds) =>
@@ -642,7 +685,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
/// <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
/// process the shutdown envelope.
/// </summary>
private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
private sealed class GatedEventFakeWorkerProcessLauncher(MxCommandKind? faultCommand = null) : IWorkerProcessLauncher, IDiagnosticFakeWorker
{
public const int ProcessId = 7730;
@@ -655,6 +698,9 @@ public sealed class GatewayEndToEndReconnectReplayTests
/// <summary>Gets the task representing the fake worker's running background loop.</summary>
public Task WorkerTask { get; private set; } = Task.CompletedTask;
/// <inheritdoc />
public FakeWorkerDiagnostics Diagnostics { get; } = new();
/// <summary>Releases the gate so the worker emits one event.</summary>
public void AllowNextEvent() => _emitGate.Release();
@@ -674,6 +720,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
CancellationToken cancellationToken = default)
{
WorkerTask = RunWorkerAsync(request, cancellationToken);
Diagnostics.Track(WorkerTask);
return Task.FromResult(new WorkerProcessHandle(
_process,
@@ -697,14 +744,19 @@ public sealed class GatewayEndToEndReconnectReplayTests
int advisedItemHandle = 0;
int emittedCount = 0;
await using GatedWorkerEnvelopeReader incoming = new(
harness.ReadGatewayEnvelopeAsync, _emitGate, cancellationToken);
while (!cancellationToken.IsCancellationRequested)
{
// While subscribed and not stopped, emit gated events using a non-blocking peek
// at the gate so incoming envelopes (including shutdown) are never starved.
while (advisedServerHandle != 0
&& !_stopEmitting
&& await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
WorkerEnvelope? envelope = await incoming.ReadOrGateAsync(
advisedServerHandle != 0 && !_stopEmitting).ConfigureAwait(false);
if (envelope is null)
{
if (_stopEmitting)
{
continue;
}
int index = ++emittedCount;
await harness.EmitEventAsync(
MxEventFamily.OnDataChange,
@@ -721,21 +773,10 @@ public sealed class GatewayEndToEndReconnectReplayTests
};
mxEvent.OnDataChange = new OnDataChangeEvent();
}).ConfigureAwait(false);
}
WorkerEnvelope? envelope;
try
{
using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Timed out waiting for an envelope — loop back to check the gate / emit.
continue;
}
Diagnostics.Received(envelope);
if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
{
await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -749,6 +790,11 @@ public sealed class GatewayEndToEndReconnectReplayTests
}
MxCommand command = envelope.WorkerCommand.Command;
if (command.Kind == faultCommand)
{
throw new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item");
}
await harness.ReplyToCommandAsync(
envelope,
configureReply: reply => ConfigureCommandReply(reply, command.Kind),
@@ -478,24 +478,42 @@ public sealed class SessionManagerTests
[Fact]
public async Task InvokeAsync_WhenWorkerHandshakingThenReadyWithinTimeout_Succeeds()
{
FakeTimeProvider clock = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
FakeWorkerClient workerClient = new() { State = WorkerClientState.Handshaking };
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(workerReadyWaitTimeoutMs: 500));
options: CreateOptions(workerReadyWaitTimeoutMs: 500),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
Assert.Equal(SessionState.Ready, session.State);
// Flip the worker to Ready shortly after the invoke starts waiting.
_ = Task.Run(async () =>
TaskCompletionSource handshakingObserved = new(TaskCreationOptions.RunContinuationsAsynchronously);
TaskCompletionSource allowReady = new(TaskCreationOptions.RunContinuationsAsynchronously);
workerClient.OnStateRead = state =>
{
await Task.Delay(50, CancellationToken.None);
workerClient.State = WorkerClientState.Ready;
});
WorkerCommandReply reply = await manager.InvokeAsync(
if (state == WorkerClientState.Handshaking)
{
handshakingObserved.TrySetResult();
}
};
Task transition = TransitionAsync();
Task<WorkerCommandReply> invoke = manager.InvokeAsync(
session.SessionId,
CreateCommand(MxCommandKind.Ping),
CancellationToken.None);
await handshakingObserved.Task.WaitAsync(TimeSpan.FromSeconds(5));
Assert.False(invoke.IsCompleted);
Assert.Equal(0, workerClient.InvokeCount);
allowReady.SetResult();
await transition.WaitAsync(TimeSpan.FromSeconds(5));
clock.Advance(TimeSpan.FromMilliseconds(50));
WorkerCommandReply reply = await invoke.WaitAsync(TimeSpan.FromSeconds(5));
async Task TransitionAsync()
{
await allowReady.Task;
workerClient.State = WorkerClientState.Ready;
}
Assert.NotNull(reply);
Assert.Equal(1, workerClient.InvokeCount);
@@ -1406,14 +1424,28 @@ public sealed class SessionManagerTests
private sealed class FakeWorkerClient : IWorkerClient
{
private WorkerClientState _state = WorkerClientState.Ready;
/// <inheritdoc />
public string SessionId { get; init; } = "session-1";
/// <inheritdoc />
public int? ProcessId { get; init; } = 1234;
/// <summary>Gets or sets the optional state-observation gate.</summary>
public Action<WorkerClientState>? OnStateRead { get; set; }
/// <inheritdoc />
public WorkerClientState State { get; set; } = WorkerClientState.Ready;
public WorkerClientState State
{
get
{
OnStateRead?.Invoke(_state);
return _state;
}
set => _state = value;
}
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt { get; init; } = DateTimeOffset.UtcNow;
@@ -0,0 +1,88 @@
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
/// <summary>Protects bounded teardown, fault observation, and safe diagnostic output.</summary>
public sealed class FakeWorkerDiagnosticsTests
{
/// <summary>Repeated gateway logs remain bounded and never format sensitive state or messages.</summary>
[Fact]
public void Log_WhenFaultsRepeat_ReportsOnlyBoundedMetadata()
{
using FakeWorkerDiagnostics diagnostics = new();
List<string> output = [];
diagnostics.SetOutput(output.Add);
for (int i = 0; i < 32; i++)
{
diagnostics.Log(LogLevel.Warning, new EventId(1), "secret-command",
new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"),
(state, exception) => $"{state} {exception}");
}
Assert.NotEmpty(output);
Assert.InRange(output.Count, 1, 16);
Assert.All(output, line =>
{
Assert.Contains("type=WorkerFrameProtocolException", line);
Assert.Contains("code=InvalidEnvelope", line);
Assert.DoesNotContain("secret", line);
});
}
/// <summary>A late fault is observed after timeout without calling the completed output sink.</summary>
/// <returns>The asynchronous test operation.</returns>
[Fact]
public async Task DrainAsync_WhenWorkerOutlivesBudget_ObservesLateFaultWithoutLateOutput()
{
FakeWorkerDiagnostics diagnostics = new();
List<string> output = [];
diagnostics.SetOutput(output.Add);
TaskCompletionSource worker = new(TaskCreationOptions.RunContinuationsAsynchronously);
diagnostics.Track(worker.Task);
await diagnostics.DrainAsync(TimeSpan.Zero);
Assert.Contains(output, line => line.Contains("source=teardown") && line.Contains("type=TimeoutException"));
int beforeFault = output.Count;
worker.SetException(new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"));
await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
Assert.Equal(beforeFault, output.Count);
}
/// <summary>Expected cancellation does not add fault diagnostics.</summary>
/// <returns>The asynchronous test operation.</returns>
[Fact]
public async Task DrainAsync_WhenWorkerIsCancelled_DoesNotReportFault()
{
FakeWorkerDiagnostics diagnostics = new();
List<string> output = [];
diagnostics.SetOutput(output.Add);
diagnostics.Track(Task.FromCanceled(new CancellationToken(true)));
await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
Assert.Empty(output);
}
/// <summary>An unavailable output helper cannot fault the observer or replace the primary failure.</summary>
/// <returns>The asynchronous test operation.</returns>
[Fact]
public async Task DrainAsync_WhenOutputThrows_PreservesPrimaryFailure()
{
InvalidOperationException primary = new("primary");
FakeWorkerDiagnostics diagnostics = new();
diagnostics.SetOutput(_ => throw new InvalidOperationException("completed output"));
Exception? failure = await Record.ExceptionAsync(async () =>
{
try
{
throw primary;
}
finally
{
diagnostics.Track(Task.FromException(new WorkerFrameProtocolException(
WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item")));
await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
}
});
Assert.Same(primary, failure);
}
}
@@ -0,0 +1,98 @@
using System.IO.Pipelines;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
/// <summary>Protects fake worker framing while the emission gate wakes its loop.</summary>
public sealed class GatedWorkerEnvelopeReaderTests
{
/// <summary>A gate wakeup must preserve a partially consumed frame and the following frame.</summary>
/// <returns>The asynchronous test operation.</returns>
/// <param name="watchGate">Whether the emission gate is watched (false before Advise).</param>
[Theory]
[InlineData(true)]
[InlineData(false)]
public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames(bool watchGate)
{
WorkerFrameProtocolOptions options = new("segmented-session");
WorkerEnvelope first = new()
{
SessionId = "segmented-session",
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
Sequence = 1,
WorkerShutdown = new WorkerShutdown(),
};
WorkerEnvelope second = first.Clone();
second.Sequence = 2;
await using MemoryStream encoded = new();
WorkerFrameWriter writer = new(encoded, options);
await writer.WriteAsync(first);
await writer.WriteAsync(second);
byte[] bytes = encoded.ToArray();
Pipe pipe = new();
await using Stream input = pipe.Reader.AsStream();
WorkerFrameReader reader = new(input, options);
using SemaphoreSlim gate = new(0);
using CancellationTokenSource lifetime = new(TimeSpan.FromSeconds(5));
await using GatedWorkerEnvelopeReader selector = new(token => reader.ReadAsync(token).AsTask(), gate, lifetime.Token);
// The length and first payload byte are consumed before the remaining bytes exist.
await pipe.Writer.WriteAsync(bytes.AsMemory(0, 5));
Task<WorkerEnvelope?> pending = selector.ReadOrGateAsync(watchGate);
Assert.False(pending.IsCompleted);
gate.Release();
if (watchGate)
{
Assert.Null(await pending.WaitAsync(TimeSpan.FromSeconds(5)));
}
else
{
// Explicit deadline probe: pre-Advise framing must remain pending beyond the old
// 50ms cancellation budget, even with the emission gate already open.
Task oldReadBudget = Task.Delay(TimeSpan.FromMilliseconds(100), lifetime.Token);
Assert.Same(oldReadBudget, await Task.WhenAny(pending, oldReadBudget));
await oldReadBudget;
}
await pipe.Writer.WriteAsync(bytes.AsMemory(5));
await pipe.Writer.CompleteAsync();
Assert.Equal(first, watchGate ? await selector.ReadOrGateAsync() : await pending.WaitAsync(TimeSpan.FromSeconds(5)));
Assert.Equal(second, await selector.ReadOrGateAsync(watchGate));
}
/// <summary>Disposal or lifetime cancellation drains the retained read and gate wait.</summary>
/// <param name="cancelLifetime">Whether cancellation precedes disposal.</param>
/// <returns>The asynchronous test operation.</returns>
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task DisposeAsync_WithIncompleteFrame_CancelsRetainedRead(bool cancelLifetime)
{
Pipe pipe = new();
await using Stream input = pipe.Reader.AsStream();
WorkerFrameReader reader = new(input, new WorkerFrameProtocolOptions("cancel-session"));
using SemaphoreSlim gate = new(0);
using CancellationTokenSource lifetime = new();
Task<WorkerEnvelope>? frameRead = null;
GatedWorkerEnvelopeReader selector = new(
token => frameRead = reader.ReadAsync(token).AsTask(), gate, lifetime.Token);
Task<WorkerEnvelope?> pending = selector.ReadOrGateAsync();
gate.Release();
Assert.Null(await pending.WaitAsync(TimeSpan.FromSeconds(5)));
Assert.NotNull(frameRead);
Assert.False(frameRead.IsCompleted);
if (cancelLifetime)
{
await lifetime.CancelAsync();
}
await selector.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5));
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => frameRead);
await pipe.Writer.CompleteAsync();
}
}
@@ -0,0 +1,135 @@
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Workers;
namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
/// <summary>Exposes test-owned worker tasks and safe diagnostic context to fixture cleanup.</summary>
internal interface IDiagnosticFakeWorker
{
/// <summary>Gets the diagnostic collector for this launcher's workers.</summary>
FakeWorkerDiagnostics Diagnostics { get; }
}
/// <summary>Observes fake worker faults without replacing the test's primary exception.</summary>
internal sealed class FakeWorkerDiagnostics : ILoggerFactory, ILogger
{
private readonly object _sync = new();
private readonly List<Task> _observations = [];
private Action<string>? _output;
private int _remaining = 16;
private bool _closed;
private string _phase = "startup";
private MxCommandKind? _command;
private ulong? _sequence;
/// <summary>Attaches the fixture's output sink before any workers are launched.</summary>
/// <param name="output">Destination for safe, bounded metadata.</param>
public void SetOutput(Action<string>? output) => _output = output;
/// <summary>Records only structural context, never envelopes or command arguments.</summary>
/// <param name="envelope">Incoming envelope whose safe metadata is retained.</param>
public void Received(WorkerEnvelope envelope)
{
lock (_sync)
{
_phase = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand ? "command" : "shutdown";
_command = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand
? envelope.WorkerCommand.Command.Kind : null;
_sequence = envelope.Sequence;
}
}
/// <summary>Retains an observation that consumes faults even if bounded teardown times out.</summary>
/// <param name="worker">Background task launched by the fake worker.</param>
/// <param name="source">Whether the task belongs to a fake worker or fixture teardown.</param>
public void Track(Task worker, string source = "fake-worker")
{
Task observation = worker.ContinueWith(completed =>
{
// Reading Exception observes every fault; retain no unobserved continuation fault.
if (completed.Exception is { } faults)
{
foreach (Exception fault in faults.Flatten().InnerExceptions)
{
Report(source, fault);
}
}
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
_observations.Add(observation);
}
/// <summary>Drains workers under the fixture's existing time limit and closes output.</summary>
/// <param name="timeout">Existing test timeout; never extended for diagnostics.</param>
/// <returns>The bounded cleanup operation.</returns>
public async Task DrainAsync(TimeSpan timeout)
{
try
{
await Task.WhenAll(_observations).WaitAsync(timeout).ConfigureAwait(false);
}
catch (TimeoutException exception)
{
Report("teardown", exception);
}
finally
{
Dispose();
}
}
private void Report(string source, Exception exception)
{
lock (_sync)
{
if (_closed || _remaining-- <= 0)
{
return;
}
string code = exception is WorkerFrameProtocolException protocol ? protocol.ErrorCode.ToString() : "none";
string line = $"source={source} type={exception.GetType().Name} code={code} phase={_phase} command={_command} sequence={_sequence}";
try
{
_output?.Invoke(line);
}
catch (Exception)
{
// A completed output helper must never mask the primary test failure.
}
}
}
/// <inheritdoc />
public ILogger CreateLogger(string categoryName) => this;
/// <inheritdoc />
public void AddProvider(ILoggerProvider provider) => throw new NotSupportedException();
/// <inheritdoc />
public void Dispose()
{
lock (_sync)
{
_closed = true;
_output = null;
}
}
/// <inheritdoc />
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
/// <inheritdoc />
public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Warning;
/// <inheritdoc />
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
// Never invoke formatter: state and exception messages may contain item names/secrets.
if (IsEnabled(logLevel) && exception is not null)
{
Report("gateway", exception);
}
}
}
@@ -0,0 +1,69 @@
using ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
/// <summary>Coordinates a fake worker's incoming frames and event emission gate.</summary>
internal sealed class GatedWorkerEnvelopeReader(
Func<CancellationToken, Task<WorkerEnvelope>> read,
SemaphoreSlim gate,
CancellationToken cancellationToken) : IAsyncDisposable
{
private readonly CancellationTokenSource _lifetime = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
private Task<WorkerEnvelope>? _pendingRead;
private Task? _pendingGate;
/// <summary>Reads an envelope, or returns null when emission should be checked.</summary>
/// <param name="watchGate">Whether the worker is currently subscribed and emitting.</param>
/// <returns>The next complete envelope, or null for an emission signal.</returns>
public async Task<WorkerEnvelope?> ReadOrGateAsync(bool watchGate = true)
{
// A gate wakeup must never cancel or replace a partially consumed frame.
_pendingRead ??= read(_lifetime.Token);
if (watchGate)
{
_pendingGate ??= gate.WaitAsync(_lifetime.Token);
await Task.WhenAny(_pendingRead, _pendingGate).ConfigureAwait(false);
if (!_pendingRead.IsCompleted)
{
await _pendingGate.ConfigureAwait(false);
_pendingGate = null;
return null;
}
}
WorkerEnvelope envelope = await _pendingRead.ConfigureAwait(false);
_pendingRead = null;
return envelope;
}
/// <inheritdoc />
public async ValueTask DisposeAsync()
{
await _lifetime.CancelAsync().ConfigureAwait(false);
try
{
if (_pendingRead is not null)
{
await _pendingRead.ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
{
}
finally
{
try
{
if (_pendingGate is not null)
{
await _pendingGate.ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
{
}
_lifetime.Dispose();
}
}
}
@@ -203,6 +203,100 @@ public sealed class AlarmDispatcherTests
Assert.Equal("Alice Smith", consumer.LastAckOperatorFullName);
}
/// <summary>Verifies that a managed ack client takes GUID acks with full operator identity.</summary>
[Fact]
public void Acknowledge_WithManagedAckClient_RoutesGuidAckToManagedClient()
{
FakeAlarmConsumer consumer = new FakeAlarmConsumer { AcknowledgeReturn = -1 };
FakeManagedAckClient managed = new FakeManagedAckClient { AcknowledgeReturn = 0 };
using AlarmDispatcher dispatcher = new AlarmDispatcher(
consumer,
new MxAccessAlarmEventSink(new MxAccessEventQueue(), new MxAccessEventMapper()),
SessionId,
managed);
dispatcher.Subscribe(@"\\HOST\Galaxy!Area1");
Guid guid = Guid.NewGuid();
int rc = dispatcher.Acknowledge(guid, "Acked", "alice", "WS01", "CORP", "Alice Smith");
Assert.Equal(0, rc);
Assert.Equal(@"\\HOST\Galaxy!Area1", managed.LastSubscription);
Assert.Equal(guid, managed.LastAckGuid);
Assert.Equal("CORP", managed.LastAckOperatorDomain);
Assert.Equal("Alice Smith", managed.LastAckOperatorFullName);
Assert.Equal(Guid.Empty, consumer.LastAckGuid);
}
/// <summary>
/// Verifies that a failed managed-client lifecycle neither fails the alarm
/// subscription nor leaves acks routed to the broken client.
/// </summary>
[Fact]
public void Subscribe_WhenManagedAckClientFails_FallsBackToConsumerAck()
{
FakeAlarmConsumer consumer = new FakeAlarmConsumer { AcknowledgeReturn = 0 };
FakeManagedAckClient managed = new FakeManagedAckClient { ThrowOnSubscribe = true };
using AlarmDispatcher dispatcher = new AlarmDispatcher(
consumer,
new MxAccessAlarmEventSink(new MxAccessEventQueue(), new MxAccessEventMapper()),
SessionId,
managed);
dispatcher.Subscribe(@"\\HOST\Galaxy!Area1");
Guid guid = Guid.NewGuid();
int rc = dispatcher.Acknowledge(guid, "Acked", "alice", "WS01", "CORP", "Alice Smith");
Assert.Equal(@"\\HOST\Galaxy!Area1", consumer.LastSubscription);
Assert.True(managed.Disposed);
Assert.Equal(0, rc);
Assert.Equal(guid, consumer.LastAckGuid);
Assert.Equal(Guid.Empty, managed.LastAckGuid);
}
/// <summary>
/// Verifies that while the failover consumer serves the subtag standby,
/// GUID acks (synthetic GUIDs) stay on the consumer, not the managed client.
/// </summary>
[Fact]
public void Acknowledge_WhenFailoverInSubtagMode_KeepsGuidAckOnConsumer()
{
FakeAlarmConsumer primary = new FakeAlarmConsumer { ThrowOnSubscribe = true };
FakeAlarmConsumer standby = new FakeAlarmConsumer { AcknowledgeReturn = 0 };
FailoverAlarmConsumer failover = new FailoverAlarmConsumer(
primary, standby, new FailoverSettings(threshold: 1, probeIntervalSeconds: 0, stableProbes: 3));
FakeManagedAckClient managed = new FakeManagedAckClient { AcknowledgeReturn = -1 };
using AlarmDispatcher dispatcher = new AlarmDispatcher(
failover,
new MxAccessAlarmEventSink(new MxAccessEventQueue(), new MxAccessEventMapper()),
SessionId,
managed);
dispatcher.Subscribe(@"\\HOST\Galaxy!Area1");
Assert.Equal(AlarmProviderMode.Subtag, failover.Mode);
Guid guid = Guid.NewGuid();
int rc = dispatcher.Acknowledge(guid, "Acked", "alice", "WS01", "CORP", "Alice Smith");
Assert.Equal(0, rc);
Assert.Equal(guid, standby.LastAckGuid);
Assert.Equal(Guid.Empty, managed.LastAckGuid);
}
/// <summary>Verifies that disposing the dispatcher disposes the managed ack client.</summary>
[Fact]
public void Dispose_WithManagedAckClient_DisposesManagedClient()
{
FakeManagedAckClient managed = new FakeManagedAckClient();
AlarmDispatcher dispatcher = new AlarmDispatcher(
new FakeAlarmConsumer(),
new MxAccessAlarmEventSink(new MxAccessEventQueue(), new MxAccessEventMapper()),
SessionId,
managed);
dispatcher.Dispose();
Assert.True(managed.Disposed);
}
/// <summary>Verifies that acknowledge-by-name calls are forwarded to the consumer.</summary>
[Fact]
public void AcknowledgeByName_WhenInvoked_ForwardsToConsumerWithFullTuple()
@@ -470,9 +564,13 @@ public sealed class AlarmDispatcherTests
AlarmTransitionEmitted?.Invoke(this, transition);
}
/// <summary>Gets or sets a value indicating whether <see cref="Subscribe"/> throws.</summary>
public bool ThrowOnSubscribe { get; set; }
/// <inheritdoc />
public void Subscribe(string subscription)
{
if (ThrowOnSubscribe) throw new InvalidOperationException("subscribe failed");
LastSubscription = subscription;
}
@@ -533,4 +631,50 @@ public sealed class AlarmDispatcherTests
Disposed = true;
}
}
private sealed class FakeManagedAckClient : IManagedAlarmAckClient
{
/// <summary>Gets the last subscription expression.</summary>
public string? LastSubscription { get; private set; }
/// <summary>Gets the GUID from the last acknowledge call.</summary>
public Guid LastAckGuid { get; private set; }
/// <summary>Gets the operator domain from the last acknowledge call.</summary>
public string? LastAckOperatorDomain { get; private set; }
/// <summary>Gets the operator full name from the last acknowledge call.</summary>
public string? LastAckOperatorFullName { get; private set; }
/// <summary>Gets or sets the return code for acknowledge operations.</summary>
public int AcknowledgeReturn { get; set; }
/// <summary>Gets or sets a value indicating whether <see cref="Subscribe"/> throws.</summary>
public bool ThrowOnSubscribe { get; set; }
/// <summary>Gets a value indicating whether this instance has been disposed.</summary>
public bool Disposed { get; private set; }
/// <inheritdoc />
public void Subscribe(string subscription)
{
if (ThrowOnSubscribe) throw new InvalidOperationException("RegisterConsumer returned 5.");
LastSubscription = subscription;
}
/// <inheritdoc />
public int AcknowledgeByGuid(
Guid alarmGuid,
string comment,
string operatorName,
string operatorNode,
string operatorDomain,
string operatorFullName)
{
LastAckGuid = alarmGuid;
LastAckOperatorDomain = operatorDomain;
LastAckOperatorFullName = operatorFullName;
return AcknowledgeReturn;
}
/// <inheritdoc />
public void Dispose()
{
Disposed = true;
}
}
}
@@ -0,0 +1,229 @@
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.MxAccess;
using Xunit;
using Xunit.Abstractions;
namespace ZB.MOM.WW.MxGateway.Worker.Tests.Probes;
/// <summary>
/// Live dev-rig smoke test for the MANAGED GUID-ack path — the
/// <see cref="AaManagedAlarmAckConsumer"/> wired into
/// <see cref="AlarmDispatcher"/>. Where <c>AlarmsLiveSmokeTests</c> acks by
/// name through the wnwrap COM consumer, this acks by GUID through AVEVA's
/// modern <c>aaAlarmManagedClient.AlarmClient.AlarmAckByGUID</c> — the path
/// that carries operator domain + full name and is not <c>E_NOTIMPL</c>.
///
/// Flow: subscribe a wnwrap reader + a managed ack client to
/// <c>\\&lt;machine&gt;\Galaxy!DEV</c>; wait for a Raise; pull a real
/// (non-degraded) alarm GUID from the reader's snapshot; ack it by GUID via
/// the dispatcher (which routes to the managed client); then assert the ack
/// registers as a subsequent <see cref="AlarmTransitionKind.Acknowledge"/>
/// transition (and that the active snapshot shows the alarm acked).
///
/// This is the runtime confirmation the static work could not close: that an
/// authenticated GUID ack actually flips state. rc=0 alone is "accepted",
/// not "applied" — the Acknowledge transition is the real assertion. For a
/// SECURED alarm the ack must run under an operator with Operate permission;
/// run the dev-rig flip script and, if the test UDA is secured, an
/// authenticated operator session (see docs/AlarmProbeFindings.md).
///
/// Skip-gated; flip <c>Skip=null</c> on the dev rig with AVEVA +
/// aaAlarmManagedClient.dll present and the flip script running.
/// </summary>
public sealed class AlarmManagedGuidAckLiveSmokeTests
{
private static readonly string SubscriptionExpression =
$@"\\{Environment.MachineName}\Galaxy!DEV";
private static readonly TimeSpan TransitionWaitTimeout = TimeSpan.FromSeconds(20);
private static readonly TimeSpan RaiseWait = TimeSpan.FromSeconds(30);
private static readonly TimeSpan AckConfirmWait = TimeSpan.FromSeconds(30);
private const string SessionId = "alarms-managed-guid-ack-smoke";
private readonly ITestOutputHelper output;
private readonly Stopwatch elapsed = Stopwatch.StartNew();
private readonly ConcurrentQueue<string> log = new ConcurrentQueue<string>();
/// <summary>Initializes a new instance of the AlarmManagedGuidAckLiveSmokeTests class.</summary>
/// <param name="output">Test output helper for logging.</param>
public AlarmManagedGuidAckLiveSmokeTests(ITestOutputHelper output)
{
this.output = output;
}
/// <summary>Acks a live alarm by GUID through the managed client and verifies the acked transition.</summary>
[Fact(Skip = "Live dev-rig smoke test — flip Skip=null with AVEVA + aaAlarmManagedClient.dll + the alarm flip script, and an operator authorized to ack.")]
public void ManagedGuidAck_AcksActiveAlarm_RegistersAcknowledge()
{
Exception? threadException = null;
var done = new ManualResetEventSlim(false);
var thread = new Thread(() =>
{
try { RunSmoke(); }
catch (Exception ex) { threadException = ex; }
finally { done.Set(); }
});
thread.IsBackground = false;
thread.SetApartmentState(ApartmentState.STA);
thread.Start();
done.Wait();
thread.Join();
output.WriteLine($"Captured {log.Count} log line(s):");
while (log.TryDequeue(out string? line)) output.WriteLine(line);
if (threadException != null) throw threadException;
}
private void RunSmoke()
{
Log($"Subscription expression: {SubscriptionExpression}");
MxAccessEventQueue queue = new MxAccessEventQueue();
WnWrapAlarmConsumer consumer = new WnWrapAlarmConsumer(
new WNWRAPCONSUMERLib.wwAlarmConsumerClass(),
maxAlarmsPerFetch: 1024);
MxAccessAlarmEventSink sink = new MxAccessAlarmEventSink(queue, new MxAccessEventMapper());
// Construct the managed ack client directly (production selects it via
// AaManagedAlarmAckConsumer.EnvGatedFactory when MXGATEWAY_USE_MANAGED_ALARM_ACK=1).
using IManagedAlarmAckClient managedAck =
new AaManagedAlarmAckConsumer(SessionId + ".ack");
using AlarmDispatcher dispatcher =
new AlarmDispatcher(consumer, sink, SessionId, managedAck);
Log("Constructed reader + managed ack client + dispatcher.");
dispatcher.Subscribe(SubscriptionExpression);
Log("Subscribe (reader + managed ack) -> ok. Driving PollOnce from this STA...");
// 1. Wait for a Raise so there is an active, unacked alarm to ack.
OnAlarmTransitionEvent? raiseBody = null;
DateTime raiseDeadline = DateTime.UtcNow + RaiseWait;
while (DateTime.UtcNow < raiseDeadline && raiseBody is null)
{
WorkerEvent? evt = WaitForTransition(queue, TransitionWaitTimeout, "raise", consumer);
if (evt is null) break;
OnAlarmTransitionEvent body = evt.Event.OnAlarmTransition;
Log("Transition: " + DescribeTransition(body));
Assert.Equal(SessionId, evt.Event.SessionId);
if (body.TransitionKind == AlarmTransitionKind.Raise) raiseBody = body;
}
Assert.NotNull(raiseBody);
// 2. Pull a REAL alarm GUID from the reader's snapshot (alarmmgr mode).
// Skip degraded/subtag records — their GUID is synthetic and not
// valid for AlarmAckByGUID.
var records = consumer.SnapshotActiveAlarms(out _);
Log($"Snapshot records: {records.Count}");
foreach (var r in records)
Log($" rec guid={r.AlarmGuid} state={r.State} degraded={r.Degraded} ref={r.ProviderName}!{r.Group}.{r.TagName}");
MxAlarmSnapshotRecord? target = records.FirstOrDefault(r =>
!r.Degraded &&
r.AlarmGuid != Guid.Empty &&
r.State == MxAlarmStateKind.UnackAlm);
Assert.NotNull(target);
Log($"Ack target GUID={target!.AlarmGuid} ref={target.ProviderName}!{target.Group}.{target.TagName}");
// 3. Ack by GUID through the dispatcher -> managed AlarmAckByGUID, with
// the current Windows identity (domain + full name are carried
// natively on this path, unlike the wnwrap 6-arg AckByName).
string user = Environment.UserName;
string node = Environment.MachineName;
string domain = Environment.UserDomainName ?? string.Empty;
Log($"Ack identity user='{user}' node='{node}' domain='{domain}'");
int rc = dispatcher.Acknowledge(
alarmGuid: target.AlarmGuid,
ackComment: "managed-guid-ack smoke",
ackOperatorName: user,
ackOperatorNode: node,
ackOperatorDomain: domain,
ackOperatorFullName: user);
Log($"dispatcher.Acknowledge(managed GUID) -> rc={rc}");
Assert.Equal(0, rc);
// 4. The ack must APPLY, not merely be accepted: wait for an
// Acknowledge transition on the same alarm, or the snapshot showing
// it acked. rc=0 with neither is the inert-ack failure the probe saw.
bool acked = false;
DateTime confirmDeadline = DateTime.UtcNow + AckConfirmWait;
while (DateTime.UtcNow < confirmDeadline && !acked)
{
WorkerEvent? evt = WaitForTransition(queue, TransitionWaitTimeout, "post-ack", consumer);
if (evt is not null)
{
OnAlarmTransitionEvent body = evt.Event.OnAlarmTransition;
Log("Post-ack transition: " + DescribeTransition(body));
if (body.TransitionKind == AlarmTransitionKind.Acknowledge)
{
acked = true;
break;
}
}
// Also accept the active snapshot flipping to an acked state for the
// target GUID, in case the ack delta was folded without a discrete
// Acknowledge transition.
var now = consumer.SnapshotActiveAlarms(out _);
if (now.Any(r => r.AlarmGuid == target.AlarmGuid &&
(r.State == MxAlarmStateKind.AckAlm || !string.IsNullOrEmpty(r.OperatorName))))
{
Log("Snapshot shows target GUID acked (AckAlm / operator populated).");
acked = true;
}
}
Assert.True(
acked,
"Managed GUID ack returned rc=0 but no Acknowledge transition or acked snapshot was observed " +
"within the confirm window — the ack was accepted but not applied (see docs/AlarmProbeFindings.md; " +
"most likely the alarm's MxSecurityOperate classification requires an authenticated operator).");
}
private WorkerEvent? WaitForTransition(
MxAccessEventQueue queue,
TimeSpan timeout,
string label,
WnWrapAlarmConsumer consumer)
{
DateTime deadline = DateTime.UtcNow + timeout;
int pollCount = 0;
while (DateTime.UtcNow < deadline)
{
try
{
consumer.PollOnce();
pollCount++;
}
catch (Exception ex)
{
Log($"PollOnce threw on poll #{pollCount + 1}: {ex.GetType().Name}: {ex.Message}");
throw;
}
if (queue.TryDequeue(out WorkerEvent? evt) && evt is not null)
{
if (evt.Event.Family == MxEventFamily.OnAlarmTransition) return evt;
Log($"Skipped non-alarm event (family={evt.Event.Family}) while waiting for {label}.");
}
Thread.Sleep(500);
}
Log($"Timed out waiting for {label} transition after {timeout.TotalSeconds:F0}s (polls={pollCount}).");
return null;
}
private static string DescribeTransition(OnAlarmTransitionEvent body)
{
return string.Format(
"kind={0} ref='{1}' operator='{2}' comment='{3}' ts={4:o}",
body.TransitionKind, body.AlarmFullReference, body.OperatorUser, body.OperatorComment,
body.TransitionTimestamp?.ToDateTime() ?? DateTime.MinValue);
}
private void Log(string line) => log.Enqueue($"[t={elapsed.Elapsed.TotalSeconds:F3}s] {line}");
}
@@ -76,7 +76,7 @@ public sealed class WorkerPipeSession
options,
() => Process.GetCurrentProcess().Id,
new WorkerPipeSessionOptions(),
() => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)),
() => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, managedAckFactory: AaManagedAlarmAckConsumer.EnvGatedFactory)),
logger,
stream)
{
@@ -98,7 +98,7 @@ public sealed class WorkerPipeSession
options,
processIdProvider,
new WorkerPipeSessionOptions(),
() => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)),
() => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, managedAckFactory: AaManagedAlarmAckConsumer.EnvGatedFactory)),
logger: null)
{
}
@@ -1585,7 +1585,7 @@ public sealed class WorkerPipeSession
// parameterless CompleteStartupHandshakeAsync is used without a
// prior factory call.
_runtimeSession ??= new MxAccessStaSession(
(eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null));
(eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, managedAckFactory: AaManagedAlarmAckConsumer.EnvGatedFactory));
IWorkerRuntimeSession session = _runtimeSession;
try
{
@@ -0,0 +1,231 @@
using System;
using System.IO;
using System.Reflection;
using WNWRAPCONSUMERLib;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Acknowledge path over AVEVA's modern managed alarm client,
/// <c>aaAlarmManagedClient.AlarmClient</c> (the C++/CLI assembly the AVEVA
/// Alarm Control itself uses: <c>AlarmClient → CwwAlarmConsumer →
/// alarmlst.dll → alarmmgr.exe</c>). Preferred over a hand-rolled native
/// WNAL consumer because <c>AlarmAckByGUID</c> performs the
/// register → match-by-GUID → submit sequence internally — the record
/// layout and selection semantics that are provider-defined (and therefore
/// NOT safely recoverable by static RE of alarmlst.dll) stay inside AVEVA's
/// own code. See <see cref="WnalNativeAckConsumer"/> for why the native
/// acquisition path is deliberately unfinished.
///
/// The <c>AlarmClient</c> members are <c>famorasm</c> (protected-internal)
/// and mxaccessgw is not an <c>InternalsVisibleTo</c> friend, so everything
/// here is invoked by reflection. In full-trust .NET Framework 4.8 that is
/// permitted. Method shapes are confirmed from the assembly metadata:
/// .ctor() (internal)
/// int InitializeConsumer(string appName)
/// int RegisterConsumer(int hWnd, string product, string app, string version, bool)
/// int Subscribe(string expr, short fromPri, short toPri,
/// localQueryType, localSortFlags,
/// localAlarmFilterState mask, localAlarmFilterState spec)
/// int AlarmAckByGUID(System.Guid guid, string comment, string oper,
/// string node, string domain, string fullName)
/// int DeregisterConsumer() / int UninitializeConsumer() / Dispose()
///
/// THREADING: the underlying consumer is apartment-affine (same as the
/// wnwrap COM path). Construct and call this only on the worker STA
/// (marshal via <c>StaRuntime.InvokeAsync</c>).
///
/// x86 ONLY — aaAlarmManagedClient.dll is a 32-bit mixed-mode PE.
///
/// OPEN (confirm at runtime, flagged inline): the trailing <c>bool</c> of
/// <c>RegisterConsumer</c> (passed <c>false</c> here); whether
/// <c>Subscribe</c> is actually required before an ack by GUID; and the
/// native status returned on a real secured ack.
/// </summary>
public sealed class AaManagedAlarmAckConsumer : IManagedAlarmAckClient
{
/// <summary>Default install location of the managed alarm client (eac.dll folder).</summary>
public const string DefaultAssemblyPath =
@"C:\Program Files (x86)\ArchestrA\Framework\Bin\Invensys_Systems__Inc__eac.dll\aaAlarmManagedClient.dll";
/// <summary>
/// Env var that opts the worker into the managed ack path. Unset/anything
/// other than "1" keeps the existing wnwrap ack behavior.
/// </summary>
public const string EnableEnvVar = "MXGATEWAY_USE_MANAGED_ALARM_ACK";
private const string AckApplicationName = "ZB.MOM.WW.MxGateway.ack";
/// <summary>
/// Production factory for the <see cref="AlarmCommandHandler"/> seam:
/// returns a managed ack client only when <see cref="EnableEnvVar"/> == "1",
/// otherwise <see langword="null"/> (ack stays on the wnwrap path). Never
/// throws — a construction failure (missing DLL, etc.) degrades to the
/// wnwrap path rather than breaking the alarm subscription.
/// </summary>
public static IManagedAlarmAckClient? EnvGatedFactory()
{
if (!string.Equals(Environment.GetEnvironmentVariable(EnableEnvVar), "1", StringComparison.Ordinal))
return null;
try
{
return new AaManagedAlarmAckConsumer(AckApplicationName);
}
catch
{
// Opt-in enhancement only: if the managed client can't load, fall
// back to the wnwrap ack path rather than failing alarm reads.
return null;
}
}
private const string TypeName = "aaAlarmManagedClient.AlarmClient";
private const BindingFlags Inst = BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic;
private readonly object syncRoot = new();
private readonly object client;
private readonly MethodInfo initializeConsumer;
private readonly MethodInfo registerConsumer;
private readonly MethodInfo subscribe;
private readonly MethodInfo alarmAckByGuid;
private readonly MethodInfo? deregisterConsumer;
private readonly MethodInfo? uninitializeConsumer;
private bool subscribed;
private bool disposed;
/// <param name="applicationName">Consumer application name (e.g. "MxGateway.ack").</param>
/// <param name="assemblyPath">Override for <see cref="DefaultAssemblyPath"/>.</param>
public AaManagedAlarmAckConsumer(string applicationName, string? assemblyPath = null)
{
if (string.IsNullOrWhiteSpace(applicationName))
throw new ArgumentException("Application name is required.", nameof(applicationName));
string path = assemblyPath ?? DefaultAssemblyPath;
if (!File.Exists(path))
throw new FileNotFoundException($"aaAlarmManagedClient.dll not found at '{path}'.", path);
Assembly asm = Assembly.LoadFrom(path);
Type type = asm.GetType(TypeName, throwOnError: true)
?? throw new InvalidOperationException($"Type '{TypeName}' not found in {path}.");
initializeConsumer = Resolve(type, "InitializeConsumer");
registerConsumer = Resolve(type, "RegisterConsumer");
subscribe = Resolve(type, "Subscribe");
alarmAckByGuid = Resolve(type, "AlarmAckByGUID");
deregisterConsumer = type.GetMethod("DeregisterConsumer", Inst, null, Type.EmptyTypes, null);
uninitializeConsumer = type.GetMethod("UninitializeConsumer", Inst, null, Type.EmptyTypes, null);
client = Activator.CreateInstance(type, nonPublic: true)
?? throw new InvalidOperationException($"Could not construct '{TypeName}'.");
ApplicationName = applicationName;
}
/// <summary>The consumer application name this instance registered under.</summary>
public string ApplicationName { get; }
private static MethodInfo Resolve(Type type, string name) =>
type.GetMethod(name, Inst)
?? throw new MissingMethodException(TypeName, name);
/// <summary>
/// Run the Initialize → Register → Subscribe lifecycle once, mirroring the
/// proven wnwrap argument values. <paramref name="subscription"/> is the
/// alarm query root, e.g. <c>\\NODE\Galaxy!Area</c>. Throws if any native
/// status is non-zero.
/// </summary>
public void Subscribe(string subscription)
{
if (subscription is null) throw new ArgumentNullException(nameof(subscription));
lock (syncRoot)
{
if (disposed) throw new ObjectDisposedException(nameof(AaManagedAlarmAckConsumer));
if (subscribed) throw new InvalidOperationException("Subscribe was already called.");
int init = (int)initializeConsumer.Invoke(client, new object[] { ApplicationName })!;
if (init != 0) throw new InvalidOperationException($"InitializeConsumer returned {init}.");
// RegisterConsumer(hWnd, product, app, version, bool). hWnd=0 => pull model.
// The trailing bool is unconfirmed; false = client/non-modal is the safe default.
int reg = (int)registerConsumer.Invoke(client, new object[]
{
0, "ZB.MOM.WW.MxGateway", ApplicationName, "1.0", false,
})!;
if (reg != 0) throw new InvalidOperationException($"RegisterConsumer returned {reg}.");
int sub = (int)subscribe.Invoke(client, BuildSubscribeArgs(subscription))!;
if (sub != 0) throw new InvalidOperationException($"Subscribe('{subscription}') returned {sub}.");
subscribed = true;
}
}
/// <summary>
/// Acknowledge a single alarm by GUID with full operator identity. Returns
/// AVEVA's native status (0 = success). A zero return means accepted, not
/// necessarily applied — verify the alarm's acked state / Historian row
/// (see docs/AlarmProbeFindings.md).
/// </summary>
public int AcknowledgeByGuid(
Guid alarmGuid,
string comment,
string operatorName,
string operatorNode,
string operatorDomain,
string operatorFullName)
{
lock (syncRoot)
{
if (disposed) throw new ObjectDisposedException(nameof(AaManagedAlarmAckConsumer));
if (!subscribed) throw new InvalidOperationException("Subscribe must be called before acknowledging.");
return (int)alarmAckByGuid.Invoke(client, new object[]
{
alarmGuid,
comment ?? string.Empty,
operatorName ?? string.Empty,
operatorNode ?? string.Empty,
operatorDomain ?? string.Empty,
operatorFullName ?? string.Empty,
})!;
}
}
/// <summary>
/// Build the 7 Subscribe args, boxing the proven wnwrap enum integers into
/// aaAlarmManagedClient's own enum parameter types (localQueryType /
/// localSortFlags / localAlarmFilterState) via <see cref="Enum.ToObject(Type,object)"/>.
/// </summary>
private object[] BuildSubscribeArgs(string subscription)
{
ParameterInfo[] p = subscribe.GetParameters();
if (p.Length != 7)
throw new InvalidOperationException($"Unexpected AlarmClient.Subscribe arity {p.Length}; expected 7.");
return new object[]
{
subscription,
(short)1, // wFromPri
(short)999, // wToPri
Enum.ToObject(p[3].ParameterType, (int)eQueryType.qtSummary),
Enum.ToObject(p[4].ParameterType, (int)eSortFlags.sfReturnNewestFirst),
Enum.ToObject(p[5].ParameterType, (int)eAlarmFilterState.asAlarmActiveNow),
Enum.ToObject(p[6].ParameterType, (int)eAlarmFilterState.asAlarmActiveNow),
};
}
/// <inheritdoc />
public void Dispose()
{
lock (syncRoot)
{
if (disposed) return;
disposed = true;
try { if (subscribed) deregisterConsumer?.Invoke(client, Array.Empty<object>()); } catch { /* best effort */ }
try { uninitializeConsumer?.Invoke(client, Array.Empty<object>()); } catch { /* best effort */ }
(client as IDisposable)?.Dispose();
}
}
}
@@ -39,6 +39,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
private readonly Action? threadAffinityCheck;
private readonly IMxAccessComObjectFactory? comFactory;
private readonly Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory;
private readonly Func<IManagedAlarmAckClient?>? managedAckFactory;
private readonly MxAccessEventMapper mapper = new MxAccessEventMapper();
private readonly object syncRoot = new object();
private AlarmDispatcher? dispatcher;
@@ -117,18 +118,29 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
/// <paramref name="comFactory"/>. Tests inject a fake so they need no
/// live COM factory.
/// </param>
/// <param name="managedAckFactory">
/// Optional per-subscription factory for a managed ack client
/// (<see cref="IManagedAlarmAckClient"/>). When it yields a non-null
/// client, GUID-keyed acknowledges route through AVEVA's modern
/// <c>AlarmClient.AlarmAckByGUID</c> instead of the wnwrap COM path.
/// Defaults to <see langword="null"/> (wnwrap ack). Production supplies
/// <see cref="AaManagedAlarmAckConsumer.EnvGatedFactory"/>, which is a
/// no-op unless <c>MXGATEWAY_USE_MANAGED_ALARM_ACK=1</c>.
/// </param>
public AlarmCommandHandler(
MxAccessEventQueue eventQueue,
Func<IMxAccessAlarmConsumer> consumerFactory,
Action? threadAffinityCheck,
IMxAccessComObjectFactory? comFactory,
Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory)
Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory,
Func<IManagedAlarmAckClient?>? managedAckFactory = null)
{
this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
this.consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
this.threadAffinityCheck = threadAffinityCheck;
this.comFactory = comFactory;
this.standbyFactory = standbyFactory;
this.managedAckFactory = managedAckFactory;
}
/// <summary>Gets a value indicating whether the handler is subscribed.</summary>
@@ -169,8 +181,14 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
failover.ProviderModeChanged += providerModeChangedHandler;
}
// Optional managed ack client (opt-in via env); null → wnwrap ack.
// A forced-subtag session never has alarmmgr GUIDs to ack, so it
// skips the extra managed consumer entirely.
IManagedAlarmAckClient? managedAck = command.ForcedMode == AlarmProviderMode.Subtag
? null
: managedAckFactory?.Invoke();
MxAccessAlarmEventSink sink = new MxAccessAlarmEventSink(eventQueue, mapper);
dispatcher = new AlarmDispatcher(consumer, sink, subscribeSessionId);
dispatcher = new AlarmDispatcher(consumer, sink, subscribeSessionId, managedAck);
try
{
dispatcher.Subscribe(subscriptionExpression);
@@ -36,20 +36,30 @@ public sealed class AlarmDispatcher : IDisposable
private readonly MxAccessAlarmEventSink sink;
private readonly string sessionId;
private readonly EventHandler<MxAlarmTransitionEvent> handler;
private IManagedAlarmAckClient? managedAck;
private bool disposed;
/// <summary>Initializes a new alarm dispatcher for the given consumer, sink, and session ID.</summary>
/// <param name="consumer">The alarm consumer.</param>
/// <param name="sink">The alarm event sink.</param>
/// <param name="sessionId">The session identifier.</param>
/// <param name="managedAck">
/// Optional managed ack client. When non-null, GUID-keyed acknowledges
/// route to it (AVEVA's modern <c>AlarmClient.AlarmAckByGUID</c>) instead
/// of the wnwrap consumer while the alarmmgr provider is active; null
/// keeps the existing wnwrap ack path. Subtag-mode acks always stay on
/// the consumer because their GUIDs are synthetic.
/// </param>
public AlarmDispatcher(
IMxAccessAlarmConsumer consumer,
MxAccessAlarmEventSink sink,
string sessionId)
string sessionId,
IManagedAlarmAckClient? managedAck = null)
{
this.consumer = consumer ?? throw new ArgumentNullException(nameof(consumer));
this.sink = sink ?? throw new ArgumentNullException(nameof(sink));
this.sessionId = sessionId ?? string.Empty;
this.managedAck = managedAck;
// Sink.Attach is the seam that propagates the session id onto the
// proto SessionId field of every emitted MxEvent. Pass the consumer
// as the "associated COM object" — sink ignores the object reference
@@ -70,8 +80,33 @@ public sealed class AlarmDispatcher : IDisposable
{
if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher));
consumer.Subscribe(subscription);
// The managed ack client maintains its own registered/subscribed
// consumer; run its lifecycle against the same expression so a later
// AlarmAckByGUID has a live consumer to act through. It is an opt-in
// enhancement, so a failed lifecycle drops back to the wnwrap ack path
// rather than failing the alarm subscription.
if (managedAck is not null)
{
try
{
managedAck.Subscribe(subscription);
}
catch
{
try { managedAck.Dispose(); } catch { /* swallow */ }
managedAck = null;
}
}
}
/// <summary>
/// True when the consumer is currently serving alarms from the subtag
/// fallback, whose alarm GUIDs are synthetic and unknown to alarmmgr.
/// </summary>
private bool IsSubtagActive() =>
consumer is SubtagAlarmConsumer ||
(consumer is FailoverAlarmConsumer failover && failover.Mode == AlarmProviderMode.Subtag);
/// <summary>
/// Forward an <c>AcknowledgeAlarm</c> request to the underlying
/// consumer's <c>AlarmAckByGUID</c>. Returns the AVEVA-native
@@ -93,6 +128,22 @@ public sealed class AlarmDispatcher : IDisposable
string ackOperatorFullName)
{
if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher));
if (managedAck is not null && !IsSubtagActive())
{
// Preferred path: AVEVA's modern AlarmClient.AlarmAckByGUID carries
// operator domain + full name and is not E_NOTIMPL like the wnwrap
// COM AlarmAckByGUID. Skipped in subtag mode: those GUIDs are
// synthetic and only the subtag consumer can resolve them (it acks
// via an AckMsg write).
return managedAck.AcknowledgeByGuid(
alarmGuid,
ackComment,
ackOperatorName,
ackOperatorNode,
ackOperatorDomain,
ackOperatorFullName);
}
return consumer.AcknowledgeByGuid(
alarmGuid,
ackComment,
@@ -268,5 +319,6 @@ public sealed class AlarmDispatcher : IDisposable
try { consumer.AlarmTransitionEmitted -= handler; } catch { /* swallow */ }
try { sink.Detach(); } catch { /* swallow */ }
try { consumer.Dispose(); } catch { /* swallow */ }
try { managedAck?.Dispose(); } catch { /* swallow */ }
}
}
@@ -0,0 +1,44 @@
using System;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Optional ack-only seam over AVEVA's modern managed alarm client
/// (<c>aaAlarmManagedClient.AlarmClient</c>). When supplied to an
/// <see cref="AlarmDispatcher"/>, GUID-keyed acknowledges route here
/// instead of the wnwrap COM consumer's <c>AlarmAckByGUID</c> (which is
/// <c>E_NOTIMPL</c> on current builds) — carrying full operator identity
/// (domain + full name) that the wnwrap 6-arg <c>AlarmAckByName</c> drops.
/// The production implementation is <see cref="AaManagedAlarmAckConsumer"/>.
/// </summary>
public interface IManagedAlarmAckClient : IDisposable
{
/// <summary>
/// Run the client's Initialize → Register → Subscribe lifecycle against
/// the same subscription expression the read consumer uses
/// (<c>\\&lt;node&gt;\Galaxy!&lt;area&gt;</c>). Must be called once,
/// on the worker STA, before <see cref="AcknowledgeByGuid"/>.
/// </summary>
/// <param name="subscription">The subscription expression.</param>
void Subscribe(string subscription);
/// <summary>
/// Acknowledge a single alarm by GUID with full operator identity.
/// Returns AVEVA's native status (0 = success; 0 means accepted, not
/// necessarily applied — verify the alarm's acked state / Historian row).
/// </summary>
/// <param name="alarmGuid">The alarm GUID.</param>
/// <param name="comment">The acknowledgment comment.</param>
/// <param name="operatorName">The operator name.</param>
/// <param name="operatorNode">The operator node.</param>
/// <param name="operatorDomain">The operator domain.</param>
/// <param name="operatorFullName">The operator full name.</param>
/// <returns>The AVEVA-native status code.</returns>
int AcknowledgeByGuid(
Guid alarmGuid,
string comment,
string operatorName,
string operatorNode,
string operatorDomain,
string operatorFullName);
}
@@ -0,0 +1,273 @@
using System;
using System.Runtime.InteropServices;
using System.Text;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Native alarm-acknowledge path over <c>alarmlst.dll</c>'s public <c>WNAL*</c>
/// C exports (the same stack the AVEVA Alarm Control uses:
/// <c>aaAlarmManagedClient → CwwAlarmConsumer → alarmlst.dll → alarmmgr.exe</c>).
///
/// Why this exists alongside <see cref="WnWrapAlarmConsumer"/>: the legacy
/// <c>WNWRAPCONSUMERLib</c> COM surface cannot carry a full/secured ack —
/// <c>AlarmAckByGUID</c> is <c>E_NOTIMPL</c>, the 8-arg <c>AlarmAckByName</c>
/// is a vendor stub (-55), and neither exposes the signed-ack password. The
/// native request record built by <see cref="WnalReqSetText"/> has six text
/// fields — node, operator, comment, <b>password</b>, domain, full name — so
/// this path can do a fully-attributed (and signed) ack.
///
/// THREADING: these calls are not inherently STA-bound the way the COM
/// consumer is, but keep all of them on the worker STA for ordering/parity
/// with the rest of the alarm surface (marshal via <c>StaRuntime.InvokeAsync</c>).
///
/// x86 ONLY: <c>alarmlst.dll</c> is 32-bit; this compiles into the x86 worker.
///
/// STATUS: the ack-SUBMIT contract (<see cref="WnalReqSetText"/>,
/// <see cref="WnalUserRequest"/>, <see cref="WnalUserRequestEx"/>) and the
/// request-record layout are confirmed by disassembly (see
/// docs/alarm-ack-findings.md in the reverse-engineering workspace). The
/// list-ACQUISITION side (obtaining <c>hAlarmList</c> + alarm handles) is
/// deliberately NOT implemented — see the note at the end of this class.
/// Nothing in the worker calls this type yet; it is kept for when a real
/// list handle can be supplied by a component that owns acquisition.
/// </summary>
public sealed class WnalNativeAckConsumer
{
private const string AlarmLst = "alarmlst.dll";
/// <summary>Request-record version stamped at WORD offset 2; WNALUserRequest rejects anything else.</summary>
private const ushort AlmAckVersion = 0x104;
/// <summary>Byte offset where WNALalarmReqSetText begins the string-offset table + heap.</summary>
private const int StringHeapBase = 0x42;
/// <summary>Sentinel the exports reject as an invalid list handle.</summary>
public const int InvalidAlarmList = -1;
// ---------------------------------------------------------------------
// CONFIRMED exports (disassembly-verified signatures). __stdcall, ANSI.
// ---------------------------------------------------------------------
/// <summary>
/// Pack operator identity + comment into an existing request record.
/// Arg order is confirmed from the debug format string. Per-field rule:
/// pass <c>null</c> to KEEP the record's existing value, "" to CLEAR it,
/// a string to SET it. Strings are single-byte ASCII (strcpy_s).
/// Return: positive = bytes used; NEGATIVE = record too small, magnitude
/// is the required <c>recSize</c>; 0 = <paramref name="lpAlReq"/> was null.
/// </summary>
[DllImport(AlarmLst, EntryPoint = "WNALalarmReqSetText",
CallingConvention = CallingConvention.StdCall, CharSet = CharSet.Ansi, ExactSpelling = true)]
private static extern int WnalReqSetText(
IntPtr lpAlReq,
[MarshalAs(UnmanagedType.LPStr)] string? nodeName,
[MarshalAs(UnmanagedType.LPStr)] string? oper,
[MarshalAs(UnmanagedType.LPStr)] string? comment,
[MarshalAs(UnmanagedType.LPStr)] string? passWd,
[MarshalAs(UnmanagedType.LPStr)] string? domainName,
[MarshalAs(UnmanagedType.LPStr)] string? fullName);
/// <summary>
/// Submit a user request (an ack) over the alarms SELECTED in
/// <paramref name="hAlarmList"/>. a2 is reserved (0); a3 is a capacity
/// hint (the native client passes the grid row count); a4 is scope
/// (0 = all, 1 = selected). Returns 0 on success (provider status),
/// non-zero on error.
/// </summary>
[DllImport(AlarmLst, EntryPoint = "WNALUserRequest",
CallingConvention = CallingConvention.StdCall, ExactSpelling = true)]
private static extern int WnalUserRequest(int hAlarmList, uint a2, uint a3, int a4, IntPtr lpAlmAck);
/// <summary>
/// Submit a user request (an ack) against an EXPLICIT array of
/// <paramref name="count"/> 4-byte alarm handles — no selection state
/// required. <paramref name="a4"/> mirrors the scope/flag scalar.
/// Returns 0 on success.
/// </summary>
[DllImport(AlarmLst, EntryPoint = "WNALUserRequestEx",
CallingConvention = CallingConvention.StdCall, ExactSpelling = true)]
private static extern int WnalUserRequestEx(
int hAlarmList, IntPtr handleArray, int count, uint a4, IntPtr lpAlmAck);
/// <summary>Acknowledge-scope values for the a4 argument.</summary>
private enum AckScope
{
All = 0,
Selected = 1,
}
/// <summary>
/// Build a request record and submit an ack for the given alarm
/// handles via <see cref="WnalUserRequestEx"/>. Caller supplies a live
/// <paramref name="hAlarmList"/> and the handle(s) obtained from the
/// list's records (acquisition is not implemented — see the class note).
/// Returns the native status (0 == success).
/// </summary>
/// <param name="hAlarmList">A registered, subscribed alarm-list handle.</param>
/// <param name="alarmHandles">The 4-byte alarm handles to acknowledge.</param>
/// <param name="comment">Ack comment.</param>
/// <param name="operatorName">Operator / user name.</param>
/// <param name="operatorNode">Operator node name.</param>
/// <param name="operatorDomain">Operator domain (carried natively, unlike the COM path).</param>
/// <param name="operatorFullName">Operator full name (carried natively, unlike the COM path).</param>
/// <param name="password">Operator password for a signed/secured ack, or null.</param>
public int AcknowledgeByHandles(
int hAlarmList,
int[] alarmHandles,
string comment,
string operatorName,
string operatorNode,
string operatorDomain,
string operatorFullName,
string? password)
{
if (hAlarmList == InvalidAlarmList)
throw new ArgumentException("Invalid alarm-list handle.", nameof(hAlarmList));
if (alarmHandles is null || alarmHandles.Length == 0)
throw new ArgumentException("At least one alarm handle is required.", nameof(alarmHandles));
IntPtr record = IntPtr.Zero;
IntPtr handles = IntPtr.Zero;
try
{
record = BuildAckRecord(
nodeName: operatorNode,
oper: operatorName,
comment: comment,
passWd: password,
domainName: operatorDomain,
fullName: operatorFullName);
handles = Marshal.AllocHGlobal(alarmHandles.Length * sizeof(int));
for (int i = 0; i < alarmHandles.Length; i++)
Marshal.WriteInt32(handles, i * sizeof(int), alarmHandles[i]);
return WnalUserRequestEx(
hAlarmList,
handles,
alarmHandles.Length,
(uint)AckScope.Selected,
record);
}
finally
{
if (handles != IntPtr.Zero) Marshal.FreeHGlobal(handles);
if (record != IntPtr.Zero) Marshal.FreeHGlobal(record);
}
}
/// <summary>
/// Acknowledge whatever is currently SELECTED in the list (a4 = Selected)
/// via <see cref="WnalUserRequest"/>. Use when selection has already been
/// established on the list rather than passing explicit handles.
/// </summary>
public int AcknowledgeSelected(
int hAlarmList,
uint rowCountHint,
string comment,
string operatorName,
string operatorNode,
string operatorDomain,
string operatorFullName,
string? password)
{
if (hAlarmList == InvalidAlarmList)
throw new ArgumentException("Invalid alarm-list handle.", nameof(hAlarmList));
IntPtr record = IntPtr.Zero;
try
{
record = BuildAckRecord(operatorNode, operatorName, comment, password, operatorDomain, operatorFullName);
return WnalUserRequest(hAlarmList, a2: 0, a3: rowCountHint, a4: (int)AckScope.Selected, record);
}
finally
{
if (record != IntPtr.Zero) Marshal.FreeHGlobal(record);
}
}
/// <summary>
/// Allocate a zeroed request record (header + room for the string heap),
/// stamp recSize + version, then let <see cref="WnalReqSetText"/> pack the
/// operator/comment text. Returns an unmanaged buffer the caller must free.
/// </summary>
private static IntPtr BuildAckRecord(
string nodeName, string oper, string comment, string? passWd, string domainName, string fullName)
{
// recSize must be >= 0x42 + sum(strlen+1 of each field). Size generously;
// SetText returns a negative "needed size" if this is too small, which we
// honor by growing once.
int capacity = EstimateCapacity(nodeName, oper, comment, passWd, domainName, fullName);
for (int attempt = 0; attempt < 2; attempt++)
{
IntPtr buf = Marshal.AllocHGlobal(capacity);
try
{
// Zero the whole record (fixed header 0x04..0x31 stays 0 — WNALUserRequest
// does not validate it; it is informational/provider-filled).
for (int off = 0; off < capacity; off += sizeof(long))
{
int remaining = capacity - off;
if (remaining >= sizeof(long)) Marshal.WriteInt64(buf, off, 0);
else for (int b = off; b < capacity; b++) Marshal.WriteByte(buf, b, 0);
}
Marshal.WriteInt16(buf, 0, (short)(ushort)capacity); // WORD recSize
Marshal.WriteInt16(buf, 2, unchecked((short)AlmAckVersion)); // WORD version = 0x104
int rc = WnalReqSetText(buf, nodeName, oper, comment, passWd, domainName, fullName);
if (rc >= 0)
return buf;
// rc < 0 => record too small; |rc| is the required size. Grow and retry.
capacity = -rc;
Marshal.FreeHGlobal(buf);
}
catch
{
Marshal.FreeHGlobal(buf);
throw;
}
}
throw new InvalidOperationException("WNALalarmReqSetText could not size the ack record.");
}
private static int EstimateCapacity(params string?[] fields)
{
int bytes = StringHeapBase;
foreach (string? f in fields)
bytes += (f is null ? 0 : Encoding.ASCII.GetByteCount(f)) + 1;
// Round up with headroom; records are tiny, over-allocation is cheap.
return Math.Max(1024, bytes + 64);
}
// ---------------------------------------------------------------------
// LIST ACQUISITION — intentionally NOT implemented here, by RE conclusion.
//
// The ABIs of WNALSelectEntry (func 0x1d) and WNALGetAlarmRecordEx (func
// 0x1c) are recoverable, but their SEMANTICS are not: WNALGetAlarmRecordEx
// does not define the alarm-record layout — it hands alarmmgr/the provider a
// set of opaque output buffers (four 0x41-byte string buffers, a 0x8b-byte
// buffer, scalars) that the PROVIDER fills. So the record's GUID offset and
// the 4-byte alarm handle that WNALUserRequestEx consumes are provider-defined
// and cannot be read out of alarmlst.dll; likewise WNALSelectEntry's arg
// meanings live on the provider side of MakeAlarmMgrFuncCall. Hand-rolling the
// acquisition from static RE alone would be guessing which bytes are the
// handle/GUID — a wrong guess acks the wrong alarm.
//
// Therefore DO NOT complete the native acquisition by inference. Obtain the
// hAlarmList + handles from a component that already encapsulates
// register/query/match correctly:
// * Preferred: aaAlarmManagedClient.dll AlarmClient.AlarmAckByGUID
// (int AlarmAckByGUID(Guid, comment, oper, node, domain, fullName)) — the
// Alarm Control's own code; managed signatures are exact (no ABI risk).
// It is internal, so invoke by reflection from the x86 worker.
// * Or drive ack through the existing WnWrapAlarmConsumer, which already
// owns a registered/subscribed consumer, and fix the auth gap (Stage 0).
// Keep the confirmed SUBMIT methods above for the case where a real
// hAlarmList + handles are supplied by such a component.
// ---------------------------------------------------------------------
}