Compare commits

...

13 Commits

Author SHA1 Message Date
Joseph Doherty 64dc1215fd feat(worker): attribute alarm ack transitions from wnwrap ACK fields
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 2m7s
ci / java (push) Failing after 2m17s
ci / portable (push) Failing after 2m23s
Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-19 19:18:24 -04:00
Joseph Doherty a94b502ce6 feat(contracts): carry operator credentials on alarm acknowledgement
Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-19 19:16:39 -04:00
Joseph Doherty bf4b9c6c09 docs(plan): AckMsg-as-operator alarm acknowledgement plan
Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-19 19:15:20 -04:00
Joseph Doherty 91bbdbf154 Merge branch 'docs/managed-alarm-ack-live-result' into main
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m14s
ci / windows-x86 (push) Successful in 2m23s
ci / portable (push) Failing after 2m32s
2026-10-18 15:15:58 -04:00
Joseph Doherty 8c095b9a56 Record the managed GUID alarm-ack live result
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m25s
ci / java (push) Successful in 2m18s
ci / portable (push) Failing after 2m39s
The 2026-10-18 live test on the dev rig showed the opt-in managed ack path
cannot work as built: AlarmClient.RegisterConsumer returns -65 alongside the
worker's wnwrap reader (the worker then silently falls back to wnwrap), the
managed client does not recognise the wnwrap GUID even standalone, and the
gateway's GUID arm sends empty identity fields. Replace the docs' claim that
the path works with the measured results and tell operators to leave
MXGATEWAY_USE_MANAGED_ALARM_ACK unset.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 15:11:11 -04:00
Joseph Doherty 6c82de55c6 Merge pull request 'Add opt-in managed GUID alarm-ack path' (#160) from feat/managed-guid-alarm-ack into main
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m59s
ci / portable (push) Failing after 2m21s
ci / java (push) Successful in 2m7s
2026-10-18 13:12:12 -04:00
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
28 changed files with 4340 additions and 483 deletions
+30 -13
View File
@@ -391,10 +391,13 @@ message AlarmFailoverConfig {
int32 failback_stable_probes = 3; // clean probes before switching back (>=1)
}
// Acknowledge a single alarm by its GUID. Operator identity fields are
// recorded atomically with the ack transition in the alarm-history log.
// The reply's hresult / native_status surfaces AVEVA's
// AlarmAckByGUID return code.
// Acknowledge a single alarm by its GUID. The worker resolves the GUID to the
// alarm's Galaxy reference from its active-alarm snapshot and acknowledges it
// by writing `comment` to `<TAGNAME>.AckMsg` over MXAccess after
// AuthenticateUser as the operator, so the alarm history records the real
// operator. The reply's hresult / native_status is 0 when the write was
// applied (or sent but unconfirmed), otherwise the AuthenticateUser HRESULT or
// the OnWriteComplete status detail.
message AcknowledgeAlarmCommand {
// Canonical 8-4-4-4-12 GUID string (e.g. "BCC47053-9542-4D65-BDAA-BCDEA6A32A73").
string alarm_guid = 1;
@@ -403,6 +406,10 @@ message AcknowledgeAlarmCommand {
string operator_node = 4;
string operator_domain = 5;
string operator_full_name = 6;
// Operator credential used for AuthenticateUser before the AckMsg write.
// Implementations must keep this field out of logs, metrics labels, command
// lines, and diagnostics.
string operator_password = 7;
}
// Snapshot the currently-active alarm set. Optional filter prefix scopes
@@ -414,9 +421,7 @@ message QueryActiveAlarmsCommand {
// Acknowledge a single alarm by its (name, provider, group) tuple. Used
// when the public RPC's AlarmFullReference (Provider!Group.Tag) cannot
// be resolved to a GUID directly. The worker invokes
// wwAlarmConsumerClass.AlarmAckByName which reaches the same alarm
// history path as AlarmAckByGUID.
// be resolved to a GUID directly. The worker acknowledges by writing `comment` to `<alarm_name>.AckMsg` (Galaxy provider only) after AuthenticateUser as the operator.
message AcknowledgeAlarmByNameCommand {
// Tag/alarm name (e.g. "TestMachine_001.TestAlarm001"). Tag itself
// may contain dots; the gateway-side parser splits on the first dot
@@ -431,6 +436,10 @@ message AcknowledgeAlarmByNameCommand {
string operator_node = 6;
string operator_domain = 7;
string operator_full_name = 8;
// Operator credential used for AuthenticateUser before the AckMsg write.
// Implementations must keep this field out of logs, metrics labels, command
// lines, and diagnostics.
string operator_password = 9;
}
message UnsubscribeBulkCommand {
@@ -725,11 +734,11 @@ message DrainEventsReply {
// Reply payload for AcknowledgeAlarmCommand AND
// AcknowledgeAlarmByNameCommand — both ack command kinds reuse this
// payload case (`MxCommandReply.acknowledge_alarm`); there is no
// dedicated by-name reply case. Surfaces AVEVA's native ack return
// code (AlarmAckByGUID for the GUID arm, AlarmAckByName for the
// by-name arm); 0 means success. The MxCommandReply's hresult field
// dedicated by-name reply case. Surfaces the ack outcome: 0 when the
// AckMsg write was applied or sent but unconfirmed; otherwise the
// AuthenticateUser HRESULT or the OnWriteComplete status detail. The MxCommandReply's hresult field
// carries the same value and is preferred for protocol consumers —
// this payload exists so the gateway-side WorkerAlarmRpcDispatcher
// this payload exists so GatewayAlarmMonitor
// can echo native_status into AcknowledgeAlarmReply.hresult without
// unpacking the outer envelope.
message AcknowledgeAlarmReplyPayload {
@@ -987,6 +996,15 @@ message AcknowledgeAlarmRequest {
// Operator principal performing the acknowledgement. The lmxopcua side
// resolves this from the OPC UA session prior to invoking the RPC.
string operator_user = 5;
// Operator's Galaxy (or OS) password. The gateway authenticates as
// operator_user with it before writing the acknowledgement so the alarm
// history records the real operator. May be empty for accounts without a
// password. Credential-sensitive: implementations must keep this field out
// of logs, metrics labels, command lines, and diagnostics.
string operator_password = 6;
// Optional domain for OS-authenticated galaxies. When set the gateway
// authenticates as "<operator_domain>\<operator_user>".
string operator_domain = 7;
}
message AcknowledgeAlarmReply {
@@ -997,8 +1015,7 @@ message AcknowledgeAlarmReply {
ProtocolStatus protocol_status = 3;
// Native ack return code echoed from the worker. The worker carries the
// ack outcome as a single int32 (AcknowledgeAlarmReplyPayload.native_status,
// = AlarmAckByName / AlarmAckByGUID return code; 0 = success); the gateway's
// WorkerAlarmRpcDispatcher copies that value here. This is the authoritative
// 0 = applied or sent; otherwise the AuthenticateUser HRESULT or OnWriteComplete detail); GatewayAlarmMonitor copies that value here. This is the authoritative
// ack-outcome field for the public RPC. Absent only when the worker reply
// omitted the value entirely (a protocol violation).
optional int32 hresult = 4;
+49
View File
@@ -793,6 +793,55 @@ 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 — does not work on this rig (live-tested
2026-10-18).** Setting `MXGATEWAY_USE_MANAGED_ALARM_ACK=1` in the
gateway's environment (workers inherit it) makes the worker build an
`AaManagedAlarmAckConsumer`, which drives
`aaAlarmManagedClient.AlarmClient` by reflection (its members are
non-public) and is meant to take GUID-keyed acks in place of the
wnwrap method. The live test shows it cannot do that as built, so leave
the flag off.
The test raised `TestMachine_001.TestAlarm001` with a secured write,
read its GUID from the wnwrap XML snapshot (the public API exposes no
alarm GUIDs), and acknowledged it through the `AcknowledgeAlarm` RPC
with the flag on:
| Step | Result |
|---|---|
| RPC ack by GUID, flag off (control) | `NotImplementedException`; `.Acked` stays `False` |
| RPC ack by GUID, flag on | Same `NotImplementedException`; `.Acked` `False`, `.AckMsg` empty |
| `AlarmClient.RegisterConsumer` in a process with a registered wnwrap consumer | `-65` — and wnwrap gets `-65` if the managed client registered first; a real `hWnd` does not help |
| Managed client alone in its own process, records loaded | `AlarmAckByGUID` → `-1`, `SelectByGUID` → `-1` for the wnwrap GUID in both byte orders, `AlarmAckSelected` → `-57` |
Three separate problems follow:
1. **The two libraries cannot share a process.** The worker always runs
a wnwrap reader, so the managed `RegisterConsumer` fails with `-65`.
The dispatcher is built to drop back to the wnwrap ack path on a
lifecycle failure (so alarm reads keep working), and it does so
silently — which is why the flag-on RPC is indistinguishable from the
control.
2. **The managed client does not recognise the wnwrap GUID.** `-1` from
`SelectByGUID` is "not found", not a security refusal, even with the
client holding records. Getting the client's own identifier for the
alarm is not reliable either: `GetHighPriAlarm` mostly throws the
FILETIME `ArgumentOutOfRangeException` described above. The
`MxSecurityOperate` question was therefore never reached.
3. **The gateway does not send the identity fields.** Its GUID arm fills
`OperatorNode`, `OperatorDomain` and `OperatorFullName` with empty
strings, so the domain and full name the managed method carries
would not arrive through the public RPC even if 1 and 2 were solved.
A working managed ack would need the client hosted out of process, a way
to resolve the alarm to the identifier the managed client uses, and the
identity fields plumbed through the gateway. Routing rules the worker
already applies stay correct and would carry over: subtag-mode acks
(synthetic GUIDs) stay on the subtag consumer, and by-name acks stay on
wnwrap because `AlarmClient` has no by-name ack. The skip-gated
`AlarmManagedGuidAckLiveSmokeTests` exercises the in-process
arrangement and hits problem 1.
### 5. STA / threading — resolved
The wnwrap COM is `ThreadingModel=Apartment`. The consumer's
+8
View File
@@ -446,6 +446,14 @@ behavior.
The alarm monitor is independent of client sessions: `AcknowledgeAlarm` and
`StreamAlarms` are session-less RPCs served by the monitor.
The worker also reads `MXGATEWAY_USE_MANAGED_ALARM_ACK` (inherited from the
gateway process). Setting it to `1` is meant to route GUID-keyed acknowledges
through AVEVA's managed alarm client, but a live test (2026-10-18) showed the
path cannot work as built: the managed client fails to register alongside the
worker's wnwrap reader and the worker silently falls back to wnwrap. Leave it
unset. It has no `MxGateway:*` option. See
[AlarmClientDiscovery](./AlarmClientDiscovery.md) ("Opt-in managed GUID ack").
### Alarm fallback options
The `Fallback` sub-section controls how the alarm feed selects between the
File diff suppressed because it is too large Load Diff
File diff suppressed because one or more lines are too long
File diff suppressed because it is too large Load Diff
@@ -391,10 +391,13 @@ message AlarmFailoverConfig {
int32 failback_stable_probes = 3; // clean probes before switching back (>=1)
}
// Acknowledge a single alarm by its GUID. Operator identity fields are
// recorded atomically with the ack transition in the alarm-history log.
// The reply's hresult / native_status surfaces AVEVA's
// AlarmAckByGUID return code.
// Acknowledge a single alarm by its GUID. The worker resolves the GUID to the
// alarm's Galaxy reference from its active-alarm snapshot and acknowledges it
// by writing `comment` to `<TAGNAME>.AckMsg` over MXAccess after
// AuthenticateUser as the operator, so the alarm history records the real
// operator. The reply's hresult / native_status is 0 when the write was
// applied (or sent but unconfirmed), otherwise the AuthenticateUser HRESULT or
// the OnWriteComplete status detail.
message AcknowledgeAlarmCommand {
// Canonical 8-4-4-4-12 GUID string (e.g. "BCC47053-9542-4D65-BDAA-BCDEA6A32A73").
string alarm_guid = 1;
@@ -403,6 +406,10 @@ message AcknowledgeAlarmCommand {
string operator_node = 4;
string operator_domain = 5;
string operator_full_name = 6;
// Operator credential used for AuthenticateUser before the AckMsg write.
// Implementations must keep this field out of logs, metrics labels, command
// lines, and diagnostics.
string operator_password = 7;
}
// Snapshot the currently-active alarm set. Optional filter prefix scopes
@@ -414,9 +421,7 @@ message QueryActiveAlarmsCommand {
// Acknowledge a single alarm by its (name, provider, group) tuple. Used
// when the public RPC's AlarmFullReference (Provider!Group.Tag) cannot
// be resolved to a GUID directly. The worker invokes
// wwAlarmConsumerClass.AlarmAckByName which reaches the same alarm
// history path as AlarmAckByGUID.
// be resolved to a GUID directly. The worker acknowledges by writing `comment` to `<alarm_name>.AckMsg` (Galaxy provider only) after AuthenticateUser as the operator.
message AcknowledgeAlarmByNameCommand {
// Tag/alarm name (e.g. "TestMachine_001.TestAlarm001"). Tag itself
// may contain dots; the gateway-side parser splits on the first dot
@@ -431,6 +436,10 @@ message AcknowledgeAlarmByNameCommand {
string operator_node = 6;
string operator_domain = 7;
string operator_full_name = 8;
// Operator credential used for AuthenticateUser before the AckMsg write.
// Implementations must keep this field out of logs, metrics labels, command
// lines, and diagnostics.
string operator_password = 9;
}
message UnsubscribeBulkCommand {
@@ -725,11 +734,11 @@ message DrainEventsReply {
// Reply payload for AcknowledgeAlarmCommand AND
// AcknowledgeAlarmByNameCommand — both ack command kinds reuse this
// payload case (`MxCommandReply.acknowledge_alarm`); there is no
// dedicated by-name reply case. Surfaces AVEVA's native ack return
// code (AlarmAckByGUID for the GUID arm, AlarmAckByName for the
// by-name arm); 0 means success. The MxCommandReply's hresult field
// dedicated by-name reply case. Surfaces the ack outcome: 0 when the
// AckMsg write was applied or sent but unconfirmed; otherwise the
// AuthenticateUser HRESULT or the OnWriteComplete status detail. The MxCommandReply's hresult field
// carries the same value and is preferred for protocol consumers —
// this payload exists so the gateway-side WorkerAlarmRpcDispatcher
// this payload exists so GatewayAlarmMonitor
// can echo native_status into AcknowledgeAlarmReply.hresult without
// unpacking the outer envelope.
message AcknowledgeAlarmReplyPayload {
@@ -987,6 +996,15 @@ message AcknowledgeAlarmRequest {
// Operator principal performing the acknowledgement. The lmxopcua side
// resolves this from the OPC UA session prior to invoking the RPC.
string operator_user = 5;
// Operator's Galaxy (or OS) password. The gateway authenticates as
// operator_user with it before writing the acknowledgement so the alarm
// history records the real operator. May be empty for accounts without a
// password. Credential-sensitive: implementations must keep this field out
// of logs, metrics labels, command lines, and diagnostics.
string operator_password = 6;
// Optional domain for OS-authenticated galaxies. When set the gateway
// authenticates as "<operator_domain>\<operator_user>".
string operator_domain = 7;
}
message AcknowledgeAlarmReply {
@@ -997,8 +1015,7 @@ message AcknowledgeAlarmReply {
ProtocolStatus protocol_status = 3;
// Native ack return code echoed from the worker. The worker carries the
// ack outcome as a single int32 (AcknowledgeAlarmReplyPayload.native_status,
// = AlarmAckByName / AlarmAckByGUID return code; 0 = success); the gateway's
// WorkerAlarmRpcDispatcher copies that value here. This is the authoritative
// 0 = applied or sent; otherwise the AuthenticateUser HRESULT or OnWriteComplete detail); GatewayAlarmMonitor copies that value here. This is the authoritative
// ack-outcome field for the public RPC. Absent only when the worker reply
// omitted the value entirely (a protocol violation).
optional int32 hresult = 4;
@@ -22,4 +22,10 @@ public static class WorkerLaunchEnvironment
/// <summary>Maximum alarm records requested in one current-alarm fetch.</summary>
public const string MaxAlarmsPerFetch = "MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
/// <summary>
/// MXAccess client name the worker's alarm-ack writer registers with. The
/// Historian records it as the acknowledgement's OperatorNode.
/// </summary>
public const string AlarmAckClientName = "MXGATEWAY_ALARM_ACK_CLIENT_NAME";
}
@@ -310,11 +310,47 @@ public sealed class ProtobufContractRoundTripTests
AlarmFullReference = "Tank01.Level.HiHi",
Comment = "shift handover",
OperatorUser = "operator2",
OperatorPassword = "s3cret",
OperatorDomain = "CORP",
};
var parsed = AcknowledgeAlarmRequest.Parser.ParseFrom(original.ToByteArray());
Assert.Equal(original, parsed);
Assert.Equal("s3cret", parsed.OperatorPassword);
Assert.Equal("CORP", parsed.OperatorDomain);
}
/// <summary>Verifies that the worker ack commands round-trip the operator password.</summary>
[Fact]
public void AcknowledgeAlarmCommands_RoundTripOperatorPassword()
{
MxCommand byGuid = new()
{
Kind = MxCommandKind.AcknowledgeAlarm,
AcknowledgeAlarmCommand = new AcknowledgeAlarmCommand
{
AlarmGuid = "BCC47053-9542-4D65-BDAA-BCDEA6A32A73",
OperatorUser = "alice",
OperatorDomain = "CORP",
OperatorPassword = "s3cret",
},
};
MxCommand byName = new()
{
Kind = MxCommandKind.AcknowledgeAlarmByName,
AcknowledgeAlarmByNameCommand = new AcknowledgeAlarmByNameCommand
{
AlarmName = "TestMachine_001.TestAlarm001",
ProviderName = "Galaxy",
GroupName = "TestArea",
OperatorUser = "alice",
OperatorPassword = "s3cret",
},
};
Assert.Equal("s3cret", MxCommand.Parser.ParseFrom(byGuid.ToByteArray()).AcknowledgeAlarmCommand.OperatorPassword);
Assert.Equal("s3cret", MxCommand.Parser.ParseFrom(byName.ToByteArray()).AcknowledgeAlarmByNameCommand.OperatorPassword);
}
/// <summary>Verifies that AcknowledgeAlarmReply round-trips with status, hresult, and diagnostics.</summary>
@@ -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();
}
}
}
@@ -17,6 +17,38 @@ public sealed class AlarmDispatcherTests
{
private const string SessionId = "session-001";
/// <summary>Verifies an Acknowledge transition is attributed to the ACK_* operator and comment.</summary>
[Fact]
public void OnTransition_WhenAcknowledged_UsesAckOperatorAndComment()
{
FakeAlarmConsumer consumer = new FakeAlarmConsumer();
MxAccessEventQueue queue = new MxAccessEventQueue();
MxAccessAlarmEventSink sink = new MxAccessAlarmEventSink(queue, new MxAccessEventMapper());
using AlarmDispatcher dispatcher = new AlarmDispatcher(consumer, sink, SessionId);
consumer.RaiseTransition(new MxAlarmTransitionEvent
{
PreviousState = MxAlarmStateKind.UnackAlm,
Record = new MxAlarmSnapshotRecord
{
AlarmGuid = Guid.NewGuid(),
ProviderName = "Galaxy",
Group = "TestArea",
TagName = "TestMachine_001.TestAlarm001",
State = MxAlarmStateKind.AckAlm,
AlarmComment = "Test alarm #1",
AckOperatorName = "alice",
AckComment = "checked",
},
});
Assert.True(queue.TryDequeue(out WorkerEvent? evt));
OnAlarmTransitionEvent body = evt!.Event.OnAlarmTransition;
Assert.Equal(AlarmTransitionKind.Acknowledge, body.TransitionKind);
Assert.Equal("alice", body.OperatorUser);
Assert.Equal("checked", body.OperatorComment);
}
/// <summary>Verifies that alarm transitions land in the queue with correctly mapped fields.</summary>
[Fact]
public void OnTransition_WhenAlarmTransitionRaised_LandsInQueueWithMappedFields()
@@ -203,6 +235,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 +596,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 +663,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;
}
}
}
@@ -801,4 +801,27 @@ public sealed class WnWrapAlarmConsumerXmlTests
ProviderName = "Galaxy",
};
}
[Fact]
public void ParseSnapshotXml_WithAckElements_PopulatesAckFields()
{
const string xml =
"<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"1\"><ALARM><GUID>2D8A9BE88FE04B0F8D996BB4D56620C7</GUID>" +
"<DATE>2026/10/19</DATE><TIME>18:2:53.812</TIME><GMTOFFSET>240</GMTOFFSET><DSTADJUST>0</DSTADJUST>" +
"<PROVIDER_NODE>DESKTOP-6JL3KKO</PROVIDER_NODE><PROVIDER_NAME>Galaxy</PROVIDER_NAME><GROUP>TestArea</GROUP>" +
"<TAGNAME>TestMachine_001.TestAlarm001</TAGNAME><TYPE>DSC</TYPE><VALUE>true</VALUE><LIMIT>true</LIMIT>" +
"<PRIORITY>500</PRIORITY><STATE>ACK_ALM</STATE><OPERATOR_NODE></OPERATOR_NODE><OPERATOR_NAME></OPERATOR_NAME>" +
"<ACK_DOMAIN_NAME>ArchestrA</ACK_DOMAIN_NAME><ACK_OPERATOR_NODE>mxa</ACK_OPERATOR_NODE>" +
"<ACK_OPERATOR_NAME>Administrator</ACK_OPERATOR_NAME><ALARM_COMMENT>Test alarm #1</ALARM_COMMENT>" +
"<ACK_COMMENT>mgack-authwrite 180714</ACK_COMMENT></ALARM></ALARM_RECORDS>";
Dictionary<Guid, MxAlarmSnapshotRecord> records = WnWrapAlarmConsumer.ParseSnapshotXml(xml);
MxAlarmSnapshotRecord record = Assert.Single(records.Values);
Assert.Equal("Administrator", record.AckOperatorName);
Assert.Equal("ArchestrA", record.AckOperatorDomain);
Assert.Equal("mxa", record.AckOperatorNode);
Assert.Equal("mgack-authwrite 180714", record.AckComment);
Assert.Equal("Test alarm #1", record.AlarmComment);
}
}
@@ -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,
@@ -193,6 +244,7 @@ public sealed class AlarmDispatcher : IDisposable
string fullReference = AlarmRecordTransitionMapper.ComposeFullReference(
record.ProviderName, record.Group, record.TagName);
bool useAck = kind == AlarmTransitionKind.Acknowledge && HasAckAttribution(record);
sink.EnqueueTransition(
alarmFullReference: fullReference,
@@ -202,15 +254,22 @@ public sealed class AlarmDispatcher : IDisposable
severity: record.Priority,
originalRaiseTimestampUtc: null,
transitionTimestampUtc: record.TransitionTimestampUtc,
operatorUser: record.OperatorName,
operatorComment: record.AlarmComment,
operatorUser: useAck ? record.AckOperatorName : record.OperatorName,
operatorComment: useAck ? record.AckComment : record.AlarmComment,
category: record.Group,
description: string.Empty,
degraded: record.Degraded);
}
private static bool HasAckAttribution(MxAlarmSnapshotRecord record) =>
!string.IsNullOrEmpty(record.AckOperatorName) || !string.IsNullOrEmpty(record.AckComment);
private static bool IsAcked(MxAlarmStateKind state) =>
state == MxAlarmStateKind.AckAlm || state == MxAlarmStateKind.AckRtn;
private static ActiveAlarmSnapshot MapToSnapshot(MxAlarmSnapshotRecord record, bool truncated)
{
bool useAck = IsAcked(record.State) && HasAckAttribution(record);
ActiveAlarmSnapshot snapshot = new ActiveAlarmSnapshot
{
AlarmFullReference = AlarmRecordTransitionMapper.ComposeFullReference(
@@ -219,8 +278,8 @@ public sealed class AlarmDispatcher : IDisposable
AlarmTypeName = record.Type,
CurrentState = MapConditionState(record.State),
Severity = record.Priority,
OperatorUser = record.OperatorName,
OperatorComment = record.AlarmComment,
OperatorUser = useAck ? record.AckOperatorName : record.OperatorName,
OperatorComment = useAck ? record.AckComment : record.AlarmComment,
Category = record.Group,
Description = string.Empty,
Degraded = record.Degraded,
@@ -268,5 +327,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);
}
@@ -39,6 +39,18 @@ public sealed class MxAlarmSnapshotRecord
public string OperatorName { get; set; } = string.Empty;
/// <summary>Gets or sets the alarm comment.</summary>
public string AlarmComment { get; set; } = string.Empty;
/// <summary>Gets or sets the acknowledgement comment (wnwrap <c>ACK_COMMENT</c>).</summary>
public string AckComment { get; set; } = string.Empty;
/// <summary>Gets or sets the acknowledging operator's name (wnwrap <c>ACK_OPERATOR_NAME</c>).</summary>
public string AckOperatorName { get; set; } = string.Empty;
/// <summary>Gets or sets the acknowledging operator's domain (wnwrap <c>ACK_DOMAIN_NAME</c>).</summary>
public string AckOperatorDomain { get; set; } = string.Empty;
/// <summary>Gets or sets the acknowledging node, i.e. the MXAccess client name (wnwrap <c>ACK_OPERATOR_NODE</c>).</summary>
public string AckOperatorNode { get; set; } = string.Empty;
/// <summary>
/// Gets or sets a value indicating whether this record was synthesized
/// by the subtag-provider fallback rather than emitted by the native
@@ -792,6 +792,10 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
string? operatorNode = null;
string? operatorName = null;
string? alarmComment = null;
string? ackComment = null;
string? ackOperatorName = null;
string? ackOperatorDomain = null;
string? ackOperatorNode = null;
foreach (XmlNode child in alarmNode.ChildNodes)
{
@@ -816,6 +820,10 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
case "OPERATOR_NODE": operatorNode ??= child.InnerText; break;
case "OPERATOR_NAME": operatorName ??= child.InnerText; break;
case "ALARM_COMMENT": alarmComment ??= child.InnerText; break;
case "ACK_COMMENT": ackComment ??= child.InnerText; break;
case "ACK_OPERATOR_NAME": ackOperatorName ??= child.InnerText; break;
case "ACK_DOMAIN_NAME": ackOperatorDomain ??= child.InnerText; break;
case "ACK_OPERATOR_NODE": ackOperatorNode ??= child.InnerText; break;
}
}
@@ -843,6 +851,10 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
OperatorNode = operatorNode ?? string.Empty,
OperatorName = operatorName ?? string.Empty,
AlarmComment = alarmComment ?? string.Empty,
AckComment = ackComment ?? string.Empty,
AckOperatorName = ackOperatorName ?? string.Empty,
AckOperatorDomain = ackOperatorDomain ?? string.Empty,
AckOperatorNode = ackOperatorNode ?? string.Empty,
};
}
return records;
@@ -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.
// ---------------------------------------------------------------------
}