Compare commits

..

9 Commits

Author SHA1 Message Date
Joseph Doherty 02228b2717 Merge branch 'main' into feat/ackmsg-as-operator
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m7s
ci / java (pull_request) Successful in 2m18s
ci / nightly-windev (pull_request) Has been skipped
ci / portable (push) Failing after 4m47s
ci / windows-x86 (pull_request) Successful in 2m40s
ci / portable (pull_request) Failing after 4m43s
ci / windows-x86 (push) Successful in 5m38s
Picks up the System.Security.Cryptography.Pkcs 10.0.13 pin (NU1903).

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:28:45 -04:00
Joseph Doherty fb20997dc9 Merge branch 'fix/pkcs-cve-pin' into main
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m47s
ci / java (push) Successful in 2m6s
ci / portable (push) Successful in 8m57s
Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:24:10 -04:00
Joseph Doherty 463748e395 fix(worker): classify ack interop failures at the COM call; ignore late completions
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m24s
ci / java (push) Successful in 2m32s
ci / nightly-windev (pull_request) Has been skipped
ci / java (pull_request) Successful in 2m38s
ci / portable (push) Failing after 2m52s
ci / portable (pull_request) Failing after 2m53s
ci / windows-x86 (pull_request) Successful in 8m8s
Address review of the #164-#166 fixes:

- The HRESULT check could not tell interop from programming errors
  (a managed ArgumentException already defaults to 0x80070057). Wrap each
  MXAccess call (AuthenticateUser, AddItem, Advise, Write) in a narrow
  filter that converts an interop failure to a private marker exception,
  and catch only that marker, so exceptions from the writer's own code
  (argument checks, caching, the pump) propagate.
- A late OnWriteComplete for a dropped handle recreated the cache entry
  that DropItem removed. The writer now records completions only for item
  handles it holds on its live registration.
- The item-handle-reuse probe now asserts that handles within one
  registration strictly increase, so a rerun is a regression check.

Refs #164, #165, #166

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:18:02 -04:00
Joseph Doherty bd82036d5f docs(worker): record live item-handle-reuse probe for ack writer
ci / windows-x86 (push) Successful in 1m29s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m47s
ci / nightly-windev (pull_request) Has been skipped
ci / portable (push) Failing after 3m1s
ci / portable (pull_request) Failing after 1m46s
ci / windows-x86 (pull_request) Successful in 1m24s
ci / java (pull_request) Successful in 2m36s
Add the Skip-gated ItemHandleReuseProbeTests probe and its windev result
(2026-10-20): within one MXAccess registration, item handles increase
monotonically and are never reused across remove/re-add, so the ack
writer's Unconfirmed drop-and-re-add mitigation holds and no per-handle
generation counter is needed. Numbering restarts only on a new
registration, which teardown already isolates. Document the guarantee at
the Dirty re-add and in AlarmProbeFindings.md.

Fixes #166

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:13:31 -04:00
Joseph Doherty f13950e453 fix(worker): forget write-completion entries for dropped ack item handles
Add MxAccessWriteCompletionCache.Remove and RemoveServer and call them
from AlarmAckWriter when it drops an item (dirty re-add, LRU eviction,
failed AddItem/Advise/Write) and when it tears down its registration, so
the cache no longer keeps one entry per dropped handle for the life of
the worker.

Fixes #165

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:11:37 -04:00
Joseph Doherty 219611c821 fix(worker): narrow AlarmAckWriter ArgumentException interop filter
Resolve the operator user, password and comment before the interop try
blocks, and match ArgumentException as an MXAccess failure only when it
carries E_INVALIDARG and is not ArgumentNullException or
ArgumentOutOfRangeException. A programming error inside the try blocks
now propagates instead of being reported as an authentication or write
failure.

Fixes #164

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:11:37 -04:00
Joseph Doherty a09fa0a4d6 fix: pin System.Security.Cryptography.Pkcs 10.0.13 (NU1903)
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m23s
ci / java (push) Successful in 2m19s
ci / portable (push) Successful in 10m44s
Microsoft.Data.SqlClient 6.0.2 pulls System.Security.Cryptography.Pkcs 9.0.4,
which carries GHSA-846f-f764-76mx (high) and GHSA-2fmv-rv2c-59x7 (moderate).
With TreatWarningsAsErrors the NU1902/NU1903 audit warnings failed restore.
10.0.13 patches both.

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-20 07:07:59 -04:00
Joseph Doherty 3e988ede88 docs: record gateway-level live verification of AckMsg-as-operator acks
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m21s
ci / nightly-windev (pull_request) Has been skipped
ci / java (push) Successful in 2m3s
ci / portable (push) Failing after 2m19s
ci / windows-x86 (pull_request) Successful in 1m19s
ci / java (pull_request) Successful in 2m2s
ci / portable (pull_request) Failing after 2m17s
Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-19 20:07:36 -04:00
Joseph Doherty e7f56622c1 fix: address final ackmsg-as-operator review findings
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m0s
ci / windows-x86 (push) Successful in 2m11s
ci / portable (push) Failing after 2m21s
- AlarmAckWriter: classify AuthenticateUser and AddItem/Advise/Write
  failures by HRESULT for any interop exception (ArgumentException for
  E_INVALIDARG, UnauthorizedAccessException, COMException); server-gone
  HRESULTs at AuthenticateUser tear down the registration and report
  WriteRejected.
- Bound the ack item cache with a 256-entry LRU (eviction best-effort
  UnAdvise+RemoveItem).
- After an Unconfirmed ack, drop and re-add the item before the next
  write so a late OnWriteComplete on the old handle is not credited.
- Pass StaRuntime.PumpPendingMessages to the ack writer through a
  pump-aware alarm handler factory so long completion waits refresh the
  STA watchdog heartbeat.
- Rust CLI: --operator is required for acknowledge-alarm.
- Docs: full ack status table, credential-oracle trade-off, upgrade note.

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
2026-10-19 19:58:24 -04:00
15 changed files with 908 additions and 42 deletions
+14 -1
View File
@@ -374,7 +374,9 @@ enum Command {
reference: String,
#[arg(long, default_value = "")]
comment: String,
#[arg(long, default_value = "")]
/// Galaxy account of the acknowledging operator (required; the gateway
/// rejects an empty operator so the real operator is recorded).
#[arg(long)]
operator: String,
/// Galaxy domain of the operator; empty authenticates the bare user name.
#[arg(long, default_value = "")]
@@ -2790,6 +2792,17 @@ mod tests {
assert!(parsed.is_ok(), "parse failed: {parsed:?}");
}
#[test]
fn acknowledge_alarm_requires_operator() {
let parsed = Cli::try_parse_from([
"mxgw",
"acknowledge-alarm",
"--reference",
"Tank01.Level.HiHi",
]);
assert!(parsed.is_err());
}
#[test]
fn acknowledge_alarm_requires_reference() {
let parsed = Cli::try_parse_from(["mxgw", "acknowledge-alarm"]);
+53
View File
@@ -492,3 +492,56 @@ The gateway acknowledges by authenticating the operator and writing `AckMsg`
(`docs/DesignDecisions.md`, "Alarm acknowledgement writes AckMsg as the
authenticated operator"). The managed ack consumer and the second wnwrap ack
consumer were removed.
### Gateway-level verification
The same result holds end to end through the gateway. With the feature build deployed on
the windev rig and `TestAlarm001` raised, an `AcknowledgeAlarm` RPC for
`Galaxy!TestArea.TestMachine_001.TestAlarm001` with `operator_user=Administrator` and an
empty password returned `PROTOCOL_STATUS_CODE_OK`, `hresult` 0. `.Acked` turned True,
`.AckMsg` held the comment, and the Historian logged the `ACK_ALM` row with
`Operator=Administrator`, `DomainName=ArchestrA`, `UserFullName=Administrator`,
`OperatorNode=MxGateway` (the default `AckClientName`), and `Description` = the comment.
A wrong password, or a user Galaxy security does not know, makes `AuthenticateUser`
throw `ArgumentException` (`E_INVALIDARG`, 0x80070057) rather than a `COMException`.
That is why the writer treats `ArgumentException` from the MXAccess call itself as an
interop failure alongside `COMException`, and reports the HRESULT the exception carries.
Such an ack returns `PROTOCOL_STATUS_CODE_MXACCESS_FAILURE`, the alarm stays unacked,
the Historian gains no ack row, and the password string does not appear in the gateway
log, `stdout.log`, or `stderr.log`. A non-Galaxy provider reference returns
`PROTOCOL_STATUS_CODE_INVALID_REQUEST` with `hresult` -1, and an empty `operator_user`
returns gRPC `InvalidArgument`.
The GUID arm of the ack command was not driven live. The public alarm RPCs expose only
the full reference, so clients reach the by-name arm; worker unit tests cover the GUID
arm.
## Item handle reuse
`AlarmAckWriter` correlates an ack with its `OnWriteComplete` by the
(server handle, item handle) pair, because the callback carries no per-write id.
After an unconfirmed ack it drops the `AckMsg` item and re-adds it so the next
ack waits on a new handle that a late completion for the old one cannot match.
That only works if MXAccess does not hand the same number out again.
`ItemHandleReuseProbeTests` (Skip-gated, worker tests) checked this on windev
(`10.100.0.48`, Galaxy `DEV`) on 2026-10-20 against
`TestMachine_001.TestAlarm001.AckMsg`, without writing:
| Sequence on one registration (server handle 1) | Item handles returned |
|---|---|
| `AddItem` + `Advise`, then `UnAdvise` + `RemoveItem`, five times | 1, 2, 3, 4, 5 |
| `AddItem`, then `RemoveItem`, five times | 6, 7, 8, 9, 10 |
| Add and remove `AckMsg`, add `Acked`, add `AckMsg` again | 11, then 12 (`Acked`), 13 |
| Add `AckMsg` twice while both are held | 14, 15 |
Within a registration the numbers increase and a freed number is never reused,
so the drop-and-re-add mitigation holds. Numbering is per registration:
`Unregister` + `Register` on the same COM object returned server handle 2 and
item handle 1, and a new `LMXProxyServerClass` returned server handle 1 and
item handle 1 again. The writer is not exposed to that restart because its
teardown detaches its `OnWriteComplete` handler from the old COM object and
clears that server handle's completion entries before it creates a new one.
No per-handle generation counter is needed.
+1 -1
View File
@@ -125,7 +125,7 @@ public string ResolveRequiredScope(object request)
}
```
The `_ => GatewayScopes.Admin` fallback is intentional: any future request type that the resolver does not recognize fails closed, requiring the strongest scope until the resolver is updated. `AcknowledgeAlarm` is treated as a write — it mutates alarm state, mirroring `MxCommandKind.Write*`. It stays `invoke:write` rather than `invoke:secure` because the request carries the operator's own Galaxy credential, which the worker authenticates before the `AckMsg` write; the credential, not the API-key scope, authorizes the identity the Historian records. `StreamAlarms` and `QueryActiveAlarms` share the alarm/event surface with `StreamEvents` and `MxCommandKind.DrainEvents`, so they carry `events:read` (the active-alarm snapshot is the same data reachable through the event surface). All three alarm RPCs are session-less: the scope check is the only authorization gate, since there is no per-session ownership to enforce.
The `_ => GatewayScopes.Admin` fallback is intentional: any future request type that the resolver does not recognize fails closed, requiring the strongest scope until the resolver is updated. `AcknowledgeAlarm` is treated as a write — it mutates alarm state, mirroring `MxCommandKind.Write*`. It stays `invoke:write` rather than `invoke:secure` because the request carries the operator's own Galaxy credential, which the worker authenticates before the `AckMsg` write; the credential, not the API-key scope, authorizes the identity the Historian records. The trade-off is that an `invoke:write` key can tell `AuthenticationFailed` (an `MXACCESS_FAILURE` whose diagnostic reports operator authentication failure) apart from every other outcome, which makes `AcknowledgeAlarm` a Galaxy-password oracle for that key — whereas the `AuthenticateUser` command itself requires `invoke:secure`. What bounds it: the caller must already hold a valid, admin-issued, revocable `invoke:write` key, and every acknowledgement is serialized through the single monitor worker session's STA, so guesses run one at a time. The gateway does **not** throttle failed operator authentications: the `ApiKeyFailureLimiter` counts only failed API-key verifications, and the ASP.NET Core rate limiter covers only the dashboard `/login` endpoint. Any lockout after repeated bad passwords comes from the Galaxy security configuration, not from the gateway. Grant `invoke:write` only to clients trusted with that exposure. `StreamAlarms` and `QueryActiveAlarms` share the alarm/event surface with `StreamEvents` and `MxCommandKind.DrainEvents`, so they carry `events:read` (the active-alarm snapshot is the same data reachable through the event surface). All three alarm RPCs are session-less: the scope check is the only authorization gate, since there is no per-session ownership to enforce.
`MxCommandRequest` is special because it multiplexes many MxAccess operations through a single RPC. The resolver inspects the embedded `MxCommandKind` so each operation gets its own scope:
+8
View File
@@ -818,6 +818,14 @@ per-request credential is workable. See
- The managed ack consumer, the WNAL native ack consumer, the second wnwrap
ack-only consumer and the opt-in managed-ack environment flag are removed.
**Upgrade note.** This is a breaking change for callers. They must now send
`operator_user` (required; empty is rejected with `InvalidArgument`) and the
operator's Galaxy password; a caller that sends no password for a
password-protected account gets `MXACCESS_FAILURE` from `AuthenticateUser`
instead of an acknowledgement. OtOpcUa's `GalaxyDriver` alarm acknowledger is a
known caller that must be updated to forward the operator's credentials before
the gateway is upgraded.
## Related Documentation
- [Gateway Process Detailed Design](./GatewayProcessDesign.md)
+8 -2
View File
@@ -110,11 +110,17 @@ The worker calls `AuthenticateUser` as the operator, then writes `operator_comme
|---|---|---|
| Write applied, or sent but no `OnWriteComplete` within the wait (unconfirmed, with a diagnostic) | `OK` | `0` |
| Unknown alarm, or no ack target (non-Galaxy provider, no ack-comment subtag) | `INVALID_REQUEST` | `-1` |
| `AuthenticateUser` failed | `MXACCESS_FAILURE` | the `AuthenticateUser` HRESULT |
| Write rejected (`OnWriteComplete` non-OK) | `MXACCESS_FAILURE` | the `detail` of the `OnWriteComplete` `MXSTATUS_PROXY` |
| `AuthenticateUser` raised an MXAccess error (bad user or wrong password surfaces as `E_INVALIDARG` `0x80070057`, raised by the interop as `ArgumentException`; `COMException` and `UnauthorizedAccessException` are classified the same way) | `MXACCESS_FAILURE` | the exception's HRESULT |
| `AuthenticateUser` returned user id `0` | `MXACCESS_FAILURE` | `-1` |
| `AuthenticateUser` failed with a server-gone HRESULT (`RPC_E_DISCONNECTED`, `RPC_S_SERVER_UNAVAILABLE`, `RPC_E_SERVERFAULT`); the writer tears down its MXAccess registration and re-registers on the next ack | `MXACCESS_FAILURE` | that HRESULT |
| `AddItem`, `Advise` or `Write` on the `AckMsg` item raised an MXAccess error (`COMException`, or `ArgumentException` for `E_INVALIDARG`); the item is dropped and re-added on the next ack, and a server-gone HRESULT also tears down the registration | `MXACCESS_FAILURE` | the exception's HRESULT |
| Write rejected (`OnWriteComplete` non-OK) | `MXACCESS_FAILURE` | the `detail` of the `OnWriteComplete` `MXSTATUS_PROXY`, or `-1` when `detail` is `0` |
| Any other exception escaping the ack (e.g. the ack writer's MXAccess `Register` failing, the monitor's worker session having no active alarm subscription — `InvalidOperationException` — or an exception from the worker's own code rather than from an MXAccess call: interop failures are classified only when the `AuthenticateUser`, `AddItem`, `Advise` or `Write` call itself raises them, and an `ArgumentNullException` or `ArgumentOutOfRangeException` is never classified because the interop does not raise them for an HRESULT) | `MXACCESS_FAILURE` | the exception's `HResult`; the diagnostic names only the exception type and HRESULT |
`reply.Statuses` carries the `OnWriteComplete` `MxStatusProxy` rows when they were observed. Because an unconfirmed write also returns `OK`, a client that must be certain should watch for the resulting transition on `StreamAlarms`. `AcknowledgeAlarm` stays under the `invoke:write` scope; see [Authorization](./Authorization.md).
**Upgrade note.** Callers must now send `operator_user` (required) and the operator's Galaxy password in `operator_password`. A caller that sends no `operator_user` gets `InvalidArgument`; a caller that sends no password for a password-protected Galaxy account gets `MXACCESS_FAILURE` from `AuthenticateUser` instead of an acknowledgement. Every existing caller must be updated before the gateway is upgraded — known callers include OtOpcUa's `GalaxyDriver` alarm acknowledger, which must supply the operator's credentials.
### `StreamAlarms`
`StreamAlarms` is a server-streaming, **session-less** RPC that attaches to the gateway's central alarm feed. The handler delegates to `IGatewayAlarmService.StreamAsync`. The stream opens with a `provider_status` and a `snapshot_status` `AlarmFeedMessage` (the current provider mode and snapshot-completeness verdict), then one `AlarmFeedMessage` carrying an `active_alarm` per currently-active alarm (the ConditionRefresh snapshot), then a single `snapshot_complete`, then a `transition` for every subsequent raise / acknowledge / clear — interleaved with a further `provider_status` on each failover/failback and a further `snapshot_status` on each change of the truncation verdict. It is served by the always-on `GatewayAlarmMonitor`, which owns a single gateway-managed worker session and fans out to every attached client — clients no longer open a session of their own. `alarm_filter_prefix`, when set, scopes the stream to a sub-tree.
+17
View File
@@ -351,6 +351,23 @@ native client correlating by handle has the identical ambiguity). The exposure
is bounded: `statuses` is an enrichment field, and `Hresult`/`ProtocolStatus`
stay acceptance-only regardless of which completion (if any) was attributed.
**The alarm ack writer narrows the window for acknowledgements.**
`AlarmAckWriter` registers its own MXAccess server with its own
`MxAccessWriteCompletionCache` and reuses one advised `AckMsg` item per alarm
(least-recently-used, capped at 256). After an ack ends unconfirmed it marks the
item dirty, and the next ack to that alarm removes and re-adds the item so the
write waits on a fresh item handle the stale completion cannot match. This
relies on MXAccess never reusing an item handle number within one
registration, which a live probe confirmed (see "Item handle reuse" in
[Alarm Probe Findings](./AlarmProbeFindings.md)). Every
drop of an item (dirty re-add, eviction, or a failed `AddItem`/`Advise`/`Write`)
also removes that handle's completion entry, and tearing down the registration
removes every entry for its server handle, so the cache does not grow by one
entry per dropped handle for the life of the worker. The writer records a
completion only for an item handle it currently holds on its live
registration, so a late `OnWriteComplete` for a dropped handle cannot recreate
its entry.
## Live alarm watch-list replacement
An auto-mode `SubscribeAlarms` with `fallback_enabled = true` always builds a
@@ -32,6 +32,9 @@
native 2.1.11; 2.1.12 patches it. Bumping Sqlite does not clear it. -->
<PackageReference Include="SQLitePCLRaw.lib.e_sqlite3" Version="2.1.12" />
<PackageReference Include="Microsoft.Data.SqlClient" Version="6.0.2" />
<!-- Security pin: GHSA-846f-f764-76mx / GHSA-2fmv-rv2c-59x7. Microsoft.Data.SqlClient 6.0.2
pulls the vulnerable System.Security.Cryptography.Pkcs 9.0.4; 10.0.13 patches it. -->
<PackageReference Include="System.Security.Cryptography.Pkcs" Version="10.0.13" />
<PackageReference Include="Polly.Core" Version="8.6.6" />
</ItemGroup>
@@ -84,7 +84,9 @@ public sealed class AlarmAckWriterTests
public void Acknowledge_Twice_ReusesItemHandle()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice());
writer.Acknowledge("T.A.AckMsg", "two", Alice());
@@ -114,7 +116,9 @@ public sealed class AlarmAckWriterTests
public void Acknowledge_WhenCachedWriteThrows_RejectsAndReAddsNextTime()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice());
server.WriteException = new COMException("x", unchecked((int)0x80004005));
@@ -141,6 +145,265 @@ public sealed class AlarmAckWriterTests
Assert.Contains("Unregister", server.Operations);
}
[Fact]
public void Acknowledge_WhenAuthenticateUserDisconnected_TearsDownAndRejects()
{
RecordingAckServer server = new RecordingAckServer
{
AuthenticateResult = 5,
AuthenticateException = new COMException("gone pw-secret", unchecked((int)0x80010108)),
};
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
Assert.Equal(AlarmAckOutcome.WriteRejected, result.Outcome);
Assert.Equal(unchecked((int)0x80010108), result.NativeStatus);
Assert.Contains("AuthenticateUser", result.Diagnostic, StringComparison.Ordinal);
Assert.Contains("0x80010108", result.Diagnostic, StringComparison.Ordinal);
Assert.DoesNotContain("pw-secret", result.Diagnostic, StringComparison.Ordinal);
Assert.Contains("Unregister", server.Operations);
Assert.DoesNotContain(server.Operations, op => op.StartsWith("AddItem", StringComparison.Ordinal));
}
[Fact]
public void Acknowledge_WhenAuthenticateUserThrowsArgumentException_ReturnsAuthenticationFailed()
{
// Live MXAccess raises E_INVALIDARG for a bad user or wrong password, which the
// interop surfaces as ArgumentException rather than COMException.
RecordingAckServer server = new RecordingAckServer
{
AuthenticateResult = 5,
AuthenticateException = new ArgumentException("Value does not fall within the expected range. pw-secret"),
};
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
Assert.Equal(AlarmAckOutcome.AuthenticationFailed, result.Outcome);
Assert.Equal(unchecked((int)0x80070057), result.NativeStatus);
Assert.DoesNotContain("pw-secret", result.Diagnostic, StringComparison.Ordinal);
Assert.DoesNotContain(server.Operations, op => op.StartsWith("Write", StringComparison.Ordinal));
Assert.DoesNotContain("Unregister", server.Operations);
}
[Fact]
public void Acknowledge_WhenAuthenticateUserThrowsArgumentNullException_Propagates()
{
// A programming error is a bug, not an operator authentication failure.
RecordingAckServer server = new RecordingAckServer
{
AuthenticateResult = 5,
AuthenticateException = new ArgumentNullException("verifyUser"),
};
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
Assert.Throws<ArgumentNullException>(() => writer.Acknowledge("T.A.AckMsg", "one", Alice()));
}
[Fact]
public void Acknowledge_WhenOwnCodeOutsideComCallThrowsArgumentException_Propagates()
{
// The pump runs between Advise and Write. An ArgumentException from it carries
// the same default HRESULT (0x80070057) as an interop E_INVALIDARG, so only
// catching at the COM call itself keeps it from being reported as WriteRejected.
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => throw new ArgumentException("bug"), TimeSpan.FromMilliseconds(10), "MxGateway");
ArgumentException thrown = Assert.Throws<ArgumentException>(() => writer.Acknowledge("T.A.AckMsg", "one", Alice()));
Assert.Equal("bug", thrown.Message);
Assert.DoesNotContain(server.Operations, op => op.StartsWith("Write", StringComparison.Ordinal));
}
[Fact]
public void Acknowledge_WhenWriteThrowsArgumentOutOfRangeException_Propagates()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5, WriteException = new ArgumentOutOfRangeException("value") };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
Assert.Throws<ArgumentOutOfRangeException>(() => writer.Acknowledge("T.A.AckMsg", "one", Alice()));
}
[Fact]
public void Acknowledge_WhenOperatorUserNameIsNull_ThrowsBeforeAuthenticating()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
Assert.Throws<ArgumentException>(() => writer.Acknowledge("T.A.AckMsg", "one", new AlarmAckOperator { UserName = null! }));
Assert.DoesNotContain(server.Operations, op => op.StartsWith("AuthenticateUser", StringComparison.Ordinal));
}
[Fact]
public void Acknowledge_WhenWriteThrowsArgumentException_RejectsAndDropsItem()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5, WriteException = new ArgumentException("bad") };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
Assert.Equal(AlarmAckOutcome.WriteRejected, result.Outcome);
Assert.Equal(unchecked((int)0x80070057), result.NativeStatus);
Assert.Contains("Write", result.Diagnostic, StringComparison.Ordinal);
Assert.Equal(new[] { 100 }, server.RemovedItemHandles);
Assert.DoesNotContain("Unregister", server.Operations);
}
[Fact]
public void Acknowledge_PastCacheCap_EvictsLeastRecentlyUsedItem()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(1), "MxGateway", maxCachedItems: 2);
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
writer.Acknowledge("C.X.AckMsg", "3", Alice()); // handle 102, evicts A (100)
Assert.Equal(new[] { 100 }, server.RemovedItemHandles);
Assert.Equal(3, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
}
[Fact]
public void Acknowledge_ReusedItem_IsMostRecentlyUsedForEviction()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway", maxCachedItems: 2);
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
writer.Acknowledge("A.X.AckMsg", "3", Alice()); // reuse 100, now most recent
writer.Acknowledge("C.X.AckMsg", "4", Alice()); // handle 102, evicts B (101)
Assert.Equal(new[] { 101 }, server.RemovedItemHandles);
Assert.Equal(3, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
Assert.Contains("UnAdvise", server.Operations);
}
[Fact]
public void Acknowledge_AfterUnconfirmed_UsesFreshHandleSoStaleCompletionIsNotCredited()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(50), "MxGateway");
AlarmAckResult first = writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, no completion
// The first write's late OnWriteComplete (handle 100) lands while the
// second write is in flight; the second write's own completion never comes.
server.OnWrite = (handle, item) => cache.Record(handle, 100, Rows(MxStatusCategory.Ok, 0));
AlarmAckResult second = writer.Acknowledge("T.A.AckMsg", "two", Alice());
Assert.Equal(AlarmAckOutcome.Unconfirmed, first.Outcome);
Assert.Equal(AlarmAckOutcome.Unconfirmed, second.Outcome);
Assert.Contains(100, server.RemovedItemHandles);
Assert.Equal(2, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
Assert.Equal(new[] { 100, 101 }, server.WrittenItemHandles);
}
[Fact]
public void Acknowledge_AfterConfirmed_ReusesHandle()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice());
writer.Acknowledge("T.A.AckMsg", "two", Alice());
Assert.Equal(new[] { 100, 100 }, server.WrittenItemHandles);
Assert.Empty(server.RemovedItemHandles);
}
[Fact]
public void Acknowledge_AfterDirtyReAdd_ForgetsDroppedHandleCompletion()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(20), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, Unconfirmed
cache.Record(7, 100, Rows(MxStatusCategory.Ok, 0)); // its late completion
Assert.Equal(1, cache.Count);
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
AlarmAckResult second = writer.Acknowledge("T.A.AckMsg", "two", Alice()); // drops 100, adds 101
Assert.Equal(AlarmAckOutcome.Applied, second.Outcome);
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 100));
Assert.Equal(1UL, cache.CurrentVersion(7, 101));
}
[Fact]
public void LateCompletionForDroppedHandle_DoesNotRecreateCacheEntry()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(20), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, Unconfirmed
server.OnWrite = (handle, item) => writer.RecordWriteCompletion(handle, item, Rows(MxStatusCategory.Ok, 0));
writer.Acknowledge("T.A.AckMsg", "two", Alice()); // drops 100, adds 101, applied
writer.RecordWriteCompletion(7, 100, Rows(MxStatusCategory.Ok, 0)); // 100's late completion
writer.RecordWriteCompletion(8, 101, Rows(MxStatusCategory.Ok, 0)); // another registration
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 100));
Assert.Equal(1UL, cache.CurrentVersion(7, 101));
}
[Fact]
public void Acknowledge_PastCacheCap_ForgetsEvictedHandleCompletion()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway", maxCachedItems: 2);
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
Assert.Equal(2, cache.Count);
writer.Acknowledge("C.X.AckMsg", "3", Alice()); // handle 102, evicts A (100)
Assert.Equal(2, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 100));
}
[Fact]
public void Dispose_ForgetsServerCompletions()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("A.X.AckMsg", "1", Alice());
writer.Acknowledge("B.X.AckMsg", "2", Alice());
cache.Record(8, 100, Rows(MxStatusCategory.Ok, 0)); // another server's entry survives
writer.Dispose();
Assert.Equal(1, cache.Count);
Assert.Equal(1UL, cache.CurrentVersion(8, 100));
}
[Fact]
public void Acknowledge_WhenWriteDisconnected_ForgetsServerCompletions()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100 recorded
server.WriteException = new COMException("x", unchecked((int)0x80010108));
writer.Acknowledge("A.X.AckMsg", "2", Alice());
Assert.Contains("Unregister", server.Operations);
Assert.Equal(0, cache.Count);
}
[Fact]
public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne()
{
@@ -180,6 +443,10 @@ internal sealed class RecordingAckServer : IMxAccessServer
public Action<int, int>? OnWrite { get; set; }
public List<int> RemovedItemHandles { get; } = new();
public List<int> WrittenItemHandles { get; } = new();
public int Register(string clientName)
{
Operations.Add("Register:" + clientName);
@@ -196,7 +463,11 @@ internal sealed class RecordingAckServer : IMxAccessServer
public int AddItem2(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition);
public void RemoveItem(int serverHandle, int itemHandle) => Operations.Add("RemoveItem");
public void RemoveItem(int serverHandle, int itemHandle)
{
Operations.Add("RemoveItem");
RemovedItemHandles.Add(itemHandle);
}
public void Advise(int serverHandle, int itemHandle)
{
@@ -216,6 +487,7 @@ internal sealed class RecordingAckServer : IMxAccessServer
public void Write(int serverHandle, int itemHandle, object? value, int userId)
{
Operations.Add("Write:" + value + ":" + userId);
WrittenItemHandles.Add(itemHandle);
if (WriteException is not null)
{
throw WriteException;
@@ -183,6 +183,39 @@ public sealed class MxAccessWriteCompletionCacheTests
Assert.Equal(77, Assert.Single(statuses).Detail);
}
/// <summary>Verifies that Remove forgets only the given handle pair.</summary>
[Fact]
public void Remove_ForgetsOnlyThatHandlePair()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, 22, BuildStatuses(detail: 2));
cache.Remove(7, 21);
cache.Remove(7, 99); // unknown pair is a no-op
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 21));
Assert.Equal(1UL, cache.CurrentVersion(7, 22));
}
/// <summary>Verifies that RemoveServer forgets every pair of that server and no other.</summary>
[Fact]
public void RemoveServer_ForgetsOnlyThatServersPairs()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, -5, BuildStatuses(detail: 2));
cache.Record(8, 21, BuildStatuses(detail: 3));
cache.Record(-7, 21, BuildStatuses(detail: 4));
cache.RemoveServer(7);
Assert.Equal(2, cache.Count);
Assert.Equal(1UL, cache.CurrentVersion(8, 21));
Assert.Equal(1UL, cache.CurrentVersion(-7, 21));
}
private static RepeatedField<MxStatusProxy> BuildStatuses(int detail)
{
return new RepeatedField<MxStatusProxy>
@@ -0,0 +1,185 @@
// ItemHandleReuseProbeTests.cs
//
// Probe for issue #166: does MXAccess hand out a new item handle number when the
// same item is removed and re-added on one LMXProxyServer registration? The
// AlarmAckWriter's Unconfirmed mitigation (drop the item, re-add it, wait on the
// fresh handle) relies on the number changing, because OnWriteComplete carries
// only the (server, item) handle pair.
//
// The probe only adds, advises and removes items. It never writes, so it does
// not acknowledge the alarm or create Historian rows.
//
// HOW TO RUN (dev rig with MXAccess installed and the DEV Galaxy running):
// 1. Remove (or set to null) the Skip parameter on the [Fact] below.
// 2. dotnet test src/ZB.MOM.WW.MxGateway.Worker.Tests/ZB.MOM.WW.MxGateway.Worker.Tests.csproj
// -p:Platform=x86 --filter FullyQualifiedName~ItemHandleReuseProbeTests
// --logger "console;verbosity=detailed"
// 3. Read the logged handles. Results are recorded in docs/AlarmProbeFindings.md.
//
// RESULT (windev, 2026-10-20): within one registration item handles increase
// monotonically and are never reused (1..15 across add/remove/re-add of the same
// item, with and without Advise, interleaved with another item, and while held
// twice). Item numbering restarts at 1 after Unregister + Register, and a fresh
// COM object is handed server handle 1 again.
using System;
using System.Collections.Generic;
using System.Runtime.InteropServices;
using System.Threading;
using ArchestrA.MxAccess;
using ZB.MOM.WW.MxGateway.Worker.MxAccess;
using Xunit.Abstractions;
namespace ZB.MOM.WW.MxGateway.Worker.Tests.Probes;
/// <summary>
/// Live probe that logs the item handles MXAccess returns when the same
/// <c>AckMsg</c> item is removed and re-added, to decide whether a late
/// <c>OnWriteComplete</c> for a dropped handle could match a re-added one.
/// </summary>
public sealed class ItemHandleReuseProbeTests
{
private const string AckItem = "TestMachine_001.TestAlarm001.AckMsg";
private const string OtherItem = "TestMachine_001.TestAlarm001.Acked";
private const int Rounds = 5;
private readonly ITestOutputHelper output;
private readonly List<string> log = new List<string>();
// Item handles returned on the first registration, in the order they were handed out.
private readonly List<int> firstRegistrationHandles = new List<int>();
/// <summary>Initializes a new instance of the <see cref="ItemHandleReuseProbeTests"/> class.</summary>
/// <param name="output">Test output helper for logging.</param>
public ItemHandleReuseProbeTests(ITestOutputHelper output)
{
this.output = output;
}
/// <summary>Adds, removes and re-adds the AckMsg item on a fresh registration and logs every handle.</summary>
[Fact(Skip = "Live dev-rig probe (issue #166) — flip Skip=null on a host with MXAccess and the DEV Galaxy running.")]
public void RemoveThenReAdd_LogsItemHandles()
{
Exception? threadException = null;
Thread thread = new Thread(() =>
{
try { RunProbe(); }
catch (Exception ex) { threadException = ex; }
});
thread.SetApartmentState(ApartmentState.STA);
thread.Start();
thread.Join();
foreach (string line in log)
{
output.WriteLine(line);
}
if (threadException is not null)
{
throw new InvalidOperationException("Probe failed: " + threadException.Message, threadException);
}
// The ack writer's drop-and-re-add mitigation depends on this: within one
// registration a handle number is never handed out twice.
Assert.Equal((2 * Rounds) + 5, firstRegistrationHandles.Count);
for (int i = 1; i < firstRegistrationHandles.Count; i++)
{
Assert.True(
firstRegistrationHandles[i] > firstRegistrationHandles[i - 1],
$"Item handle {firstRegistrationHandles[i]} did not increase past {firstRegistrationHandles[i - 1]}; MXAccess reused or reordered item handles within one registration.");
}
}
private void RunProbe()
{
LMXProxyServerClass com = new LMXProxyServerClass();
try
{
IMxAccessServer mx = new MxAccessComServer(com);
int server = mx.Register("MxGatewayProbe166");
Log($"Register -> server handle {server}");
// A: the AlarmAckWriter drop path (Advise, then UnAdvise + RemoveItem).
for (int i = 0; i < Rounds; i++)
{
int h = Held(mx.AddItem(server, AckItem));
mx.Advise(server, h);
Log($"A{i}: AddItem+Advise {AckItem} -> {h}");
mx.UnAdvise(server, h);
mx.RemoveItem(server, h);
Log($"A{i}: UnAdvise+RemoveItem {h}");
}
// B: RemoveItem without Advise.
for (int i = 0; i < Rounds; i++)
{
int h = Held(mx.AddItem(server, AckItem));
Log($"B{i}: AddItem {AckItem} -> {h}");
mx.RemoveItem(server, h);
Log($"B{i}: RemoveItem {h}");
}
// C: interleave another item to see whether a freed number is handed to any item.
int c1 = Held(mx.AddItem(server, AckItem));
Log($"C: AddItem {AckItem} -> {c1}");
mx.RemoveItem(server, c1);
Log($"C: RemoveItem {c1}");
int c2 = Held(mx.AddItem(server, OtherItem));
Log($"C: AddItem {OtherItem} -> {c2}");
int c3 = Held(mx.AddItem(server, AckItem));
Log($"C: AddItem {AckItem} -> {c3}");
mx.RemoveItem(server, c2);
mx.RemoveItem(server, c3);
Log($"C: RemoveItem {c2}, {c3}");
// D: the same item added twice while the first is still held.
int d1 = Held(mx.AddItem(server, AckItem));
int d2 = Held(mx.AddItem(server, AckItem));
Log($"D: AddItem {AckItem} twice while held -> {d1}, {d2}");
mx.RemoveItem(server, d1);
mx.RemoveItem(server, d2);
mx.Unregister(server);
Log($"Unregister {server}");
// E: the TearDownServer path: re-register on the same COM object and add again.
int server2 = mx.Register("MxGatewayProbe166");
int e1 = mx.AddItem(server2, AckItem);
Log($"E: re-Register -> server handle {server2}; AddItem {AckItem} -> {e1}");
mx.RemoveItem(server2, e1);
mx.Unregister(server2);
}
finally
{
Marshal.FinalReleaseComObject(com);
}
// F: a fresh COM object, as AlarmAckWriter creates after a teardown.
LMXProxyServerClass com2 = new LMXProxyServerClass();
try
{
IMxAccessServer mx2 = new MxAccessComServer(com2);
int server3 = mx2.Register("MxGatewayProbe166");
int f1 = mx2.AddItem(server3, AckItem);
Log($"F: new COM object Register -> server handle {server3}; AddItem {AckItem} -> {f1}");
mx2.RemoveItem(server3, f1);
mx2.Unregister(server3);
}
finally
{
Marshal.FinalReleaseComObject(com2);
}
}
private int Held(int itemHandle)
{
firstRegistrationHandles.Add(itemHandle);
return itemHandle;
}
private void Log(string line)
{
log.Add(DateTime.Now.ToString("HH:mm:ss.fff") + " " + 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, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump)),
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, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump)),
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, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump));
IWorkerRuntimeSession session = _runtimeSession;
try
{
@@ -26,13 +26,30 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
/// <summary>Default MXAccess client name when none is configured.</summary>
public const string DefaultClientName = "MxGateway";
/// <summary>
/// Default cap on cached ack item handles. Each cached item holds an
/// advised MXAccess item for the worker's life, so the cache is bounded
/// (least-recently-used eviction) rather than growing with every
/// distinct alarm ever acknowledged.
/// </summary>
public const int DefaultMaxCachedItems = 256;
private readonly IMxAccessComObjectFactory? factory;
private readonly MxAccessWriteCompletionCache completionCache;
private readonly Action pumpStep;
private readonly TimeSpan completionTimeout;
private readonly string clientName;
private readonly MxStatusProxyConverter statusConverter = new MxStatusProxyConverter();
private readonly Dictionary<string, int> itemHandles = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
private readonly int maxCachedItems;
private readonly Dictionary<string, LinkedListNode<AckItem>> itemHandles = new Dictionary<string, LinkedListNode<AckItem>>(StringComparer.OrdinalIgnoreCase);
// Most-recently-used first; the tail is the eviction candidate.
private readonly LinkedList<AckItem> itemLru = new LinkedList<AckItem>();
// Item handles currently added on this registration. A completion for any
// other handle (a dropped one whose OnWriteComplete arrives late) is ignored
// so it cannot recreate a completion-cache entry for a dead handle.
private readonly HashSet<int> heldItemHandles = new HashSet<int>();
private object? comObject;
private LMXProxyServerClass? comEventSource;
private IMxAccessServer? server;
@@ -54,6 +71,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
this.pumpStep = pumpStep ?? throw new ArgumentNullException(nameof(pumpStep));
this.completionTimeout = completionTimeout;
this.clientName = string.IsNullOrWhiteSpace(clientName) ? DefaultClientName : clientName!;
maxCachedItems = DefaultMaxCachedItems;
completionCache = new MxAccessWriteCompletionCache();
}
@@ -63,8 +81,11 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
MxAccessWriteCompletionCache completionCache,
Action pumpStep,
TimeSpan completionTimeout,
string clientName)
string clientName,
int maxCachedItems = DefaultMaxCachedItems)
{
if (maxCachedItems < 1) throw new ArgumentOutOfRangeException(nameof(maxCachedItems));
this.maxCachedItems = maxCachedItems;
this.server = server ?? throw new ArgumentNullException(nameof(server));
this.completionCache = completionCache ?? throw new ArgumentNullException(nameof(completionCache));
this.pumpStep = pumpStep ?? throw new ArgumentNullException(nameof(pumpStep));
@@ -88,19 +109,36 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
if (string.IsNullOrWhiteSpace(itemAddress)) throw new ArgumentException("Ack item address is required.", nameof(itemAddress));
if (ackOperator is null) throw new ArgumentNullException(nameof(ackOperator));
if (ackOperator.UserName is null) throw new ArgumentException("Operator user name is required.", nameof(ackOperator));
string verifyUser = ackOperator.VerifyUser;
string password = ackOperator.Password ?? string.Empty;
string commentValue = comment ?? string.Empty;
IMxAccessServer mx = EnsureRegistered();
int userId;
try
{
userId = mx.AuthenticateUser(serverHandle, ackOperator.VerifyUser, ackOperator.Password ?? string.Empty);
userId = Interop("AuthenticateUser", () => mx.AuthenticateUser(serverHandle, verifyUser, password));
}
catch (COMException ex)
catch (AckInteropException ex)
{
int authHresult = ex.HResult;
if (IsServerGone(authHresult))
{
// The registration is dead, not the credentials: tear down so the
// next ack re-registers, and report it as a write-path failure.
TearDownServer();
return AlarmAckResult.Create(
AlarmAckOutcome.WriteRejected,
authHresult,
$"AckMsg AuthenticateUser failed (HRESULT 0x{authHresult:X8}).");
}
return AlarmAckResult.Create(
AlarmAckOutcome.AuthenticationFailed,
ex.HResult,
$"Operator authentication failed for '{ackOperator.VerifyUser}' (HRESULT 0x{ex.HResult:X8}).");
$"Operator authentication failed for '{verifyUser}' (HRESULT 0x{ex.HResult:X8}).");
}
if (userId == 0)
@@ -108,36 +146,59 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
return AlarmAckResult.Create(
AlarmAckOutcome.AuthenticationFailed,
-1,
$"Operator authentication failed for '{ackOperator.VerifyUser}' (no user id returned).");
$"Operator authentication failed for '{verifyUser}' (no user id returned).");
}
ulong baselineVersion = 0;
int itemHandle = 0;
bool cached = itemHandles.TryGetValue(itemAddress, out itemHandle);
string step = "AddItem";
LinkedListNode<AckItem>? node;
bool cached = itemHandles.TryGetValue(itemAddress, out node);
try
{
if (!cached)
if (cached && node!.Value.Dirty)
{
itemHandle = mx.AddItem(serverHandle, itemAddress);
itemHandles[itemAddress] = itemHandle;
step = "Advise";
// The previous ack to this item ended Unconfirmed, so its
// OnWriteComplete may still arrive. Completions are keyed by
// (server, item) handle with no per-write id, so a late one would
// be credited to this write. Drop the item and re-add it so this
// write waits on a fresh handle the stale completion cannot match.
// That holds because MXAccess never reuses an item handle number
// within one registration: the live probe (ItemHandleReuseProbeTests,
// windev 2026-10-20) saw strictly increasing handles across repeated
// remove/re-add of the same item. Numbering restarts only on a new
// registration, and TearDownServer detaches this writer's
// OnWriteComplete handler and clears the server's completion entries
// first, so the old registration's completions cannot reach it.
DropItem(mx, node);
cached = false;
}
if (cached)
{
itemHandle = node!.Value.Handle;
itemLru.Remove(node);
itemLru.AddFirst(node);
}
else
{
itemHandle = Interop("AddItem", () => mx.AddItem(serverHandle, itemAddress));
node = CacheItem(mx, itemAddress, itemHandle);
int added = itemHandle;
// MXAccess rejects writes to added-but-unadvised items with E_INVALIDARG.
mx.Advise(serverHandle, itemHandle);
Interop("Advise", () => mx.Advise(serverHandle, added));
}
pumpStep();
step = "Write";
baselineVersion = completionCache.CurrentVersion(serverHandle, itemHandle);
mx.Write(serverHandle, itemHandle, comment ?? string.Empty, userId);
int target = itemHandle;
Interop("Write", () => mx.Write(serverHandle, target, commentValue, userId));
}
catch (COMException ex)
catch (AckInteropException ex)
{
int hresult = ex.HResult;
if (itemHandles.Remove(itemAddress))
if (itemHandles.TryGetValue(itemAddress, out LinkedListNode<AckItem>? failed))
{
try { mx.UnAdvise(serverHandle, itemHandle); } catch { /* best effort */ }
try { mx.RemoveItem(serverHandle, itemHandle); } catch { /* best effort */ }
DropItem(mx, failed!);
}
if (IsServerGone(hresult))
@@ -149,7 +210,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
return AlarmAckResult.Create(
AlarmAckOutcome.WriteRejected,
hresult,
$"AckMsg {step} failed (HRESULT 0x{hresult:X8}).");
$"AckMsg {ex.Step} failed (HRESULT 0x{hresult:X8}).");
}
ulong baseline = baselineVersion;
@@ -162,6 +223,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
pumpStep,
out RepeatedField<MxStatusProxy> statuses))
{
node!.Value.Dirty = true;
return AlarmAckResult.Create(
AlarmAckOutcome.Unconfirmed,
0,
@@ -192,6 +254,54 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
TearDownServer();
}
/// <summary>
/// Runs one MXAccess COM call and converts a failure the interop raised
/// for its HRESULT into <see cref="AckInteropException"/>. Only the COM
/// call itself runs inside the filter, so an exception thrown by this
/// writer's own code (argument checks, caching, the pump) propagates as
/// a bug instead of being reported as an MXAccess failure.
/// </summary>
/// <remarks>
/// The runtime maps well-known HRESULTs to typed exceptions and the rest
/// to COMException; each carries the HRESULT in Exception.HResult. The
/// ArgumentException arm is needed because a wrong password or unknown
/// user makes <c>AuthenticateUser</c> throw ArgumentException with
/// E_INVALIDARG (0x80070057), observed live on windev on 2026-10-19.
/// ArgumentNullException and ArgumentOutOfRangeException are excluded:
/// the interop does not raise them for an HRESULT, so one escaping the
/// call is an argument bug in the adapter, not an MXAccess failure.
/// </remarks>
private static T Interop<T>(string step, Func<T> call)
{
try
{
return call();
}
catch (Exception ex) when (IsInteropFailure(ex))
{
throw new AckInteropException(step, ex);
}
}
private static void Interop(string step, Action call)
{
try
{
call();
}
catch (Exception ex) when (IsInteropFailure(ex))
{
throw new AckInteropException(step, ex);
}
}
private static bool IsInteropFailure(Exception ex) =>
ex is ExternalException
|| ex is UnauthorizedAccessException
|| (ex is ArgumentException
&& ex is not ArgumentNullException
&& ex is not ArgumentOutOfRangeException);
private static bool IsServerGone(int hresult) =>
hresult == unchecked((int)0x80010108) // RPC_E_DISCONNECTED
|| hresult == unchecked((int)0x800706BA) // RPC_S_SERVER_UNAVAILABLE
@@ -208,16 +318,21 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
IMxAccessServer? mx = server;
if (mx is not null)
{
foreach (int itemHandle in itemHandles.Values)
foreach (AckItem item in itemLru)
{
try { mx.UnAdvise(serverHandle, itemHandle); } catch { /* best effort */ }
try { mx.RemoveItem(serverHandle, itemHandle); } catch { /* best effort */ }
try { mx.UnAdvise(serverHandle, item.Handle); } catch { /* best effort */ }
try { mx.RemoveItem(serverHandle, item.Handle); } catch { /* best effort */ }
}
try { mx.Unregister(serverHandle); } catch { /* best effort */ }
}
// The registration is gone, so no completion can arrive for its handles.
completionCache.RemoveServer(serverHandle);
itemHandles.Clear();
itemLru.Clear();
heldItemHandles.Clear();
server = null;
comEventSource = null;
object? toRelease = comObject;
@@ -228,6 +343,34 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
}
}
/// <summary>Caches a newly added item as most recently used, evicting the least recently used past the cap.</summary>
private LinkedListNode<AckItem> CacheItem(IMxAccessServer mx, string itemAddress, int itemHandle)
{
while (itemLru.Count >= maxCachedItems && itemLru.Last is { } leastRecent)
{
DropItem(mx, leastRecent);
}
LinkedListNode<AckItem> node = itemLru.AddFirst(new AckItem(itemAddress, itemHandle));
itemHandles[itemAddress] = node;
heldItemHandles.Add(itemHandle);
return node;
}
/// <summary>Removes a cached item and best-effort releases its MXAccess item.</summary>
private void DropItem(IMxAccessServer mx, LinkedListNode<AckItem> node)
{
itemHandles.Remove(node.Value.Address);
itemLru.Remove(node);
heldItemHandles.Remove(node.Value.Handle);
try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ }
try { mx.RemoveItem(serverHandle, node.Value.Handle); } catch { /* best effort */ }
// Without this the completion cache keeps one entry per dropped handle
// (each Unconfirmed re-add and each eviction) for the worker's life.
completionCache.Remove(serverHandle, node.Value.Handle);
}
private IMxAccessServer EnsureRegistered()
{
if (server is not null) return server;
@@ -274,6 +417,53 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
{
RepeatedField<MxStatusProxy> rows = new RepeatedField<MxStatusProxy>();
rows.AddRange(statusConverter.ConvertMany(pVars));
completionCache.Record(hLMXServerHandle, phItemHandle, rows);
RecordWriteCompletion(hLMXServerHandle, phItemHandle, rows);
}
/// <summary>
/// Records a write completion only for an item handle this writer
/// currently holds on its live registration. A late completion for a
/// dropped handle would otherwise recreate the cache entry that
/// <see cref="DropItem"/> removed. Internal so tests can stand in for
/// the COM event.
/// </summary>
internal void RecordWriteCompletion(int completionServerHandle, int itemHandle, RepeatedField<MxStatusProxy> rows)
{
if (server is null || completionServerHandle != serverHandle || !heldItemHandles.Contains(itemHandle))
{
return;
}
completionCache.Record(completionServerHandle, itemHandle, rows);
}
/// <summary>An MXAccess failure raised by one COM call, carrying that call's HRESULT.</summary>
private sealed class AckInteropException : Exception
{
public AckInteropException(string step, Exception inner)
: base("MXAccess " + step + " failed.", inner)
{
Step = step;
HResult = inner.HResult;
}
public string Step { get; }
}
/// <summary>A cached ack item. Plain class (not a record) for the net48 worker target.</summary>
private sealed class AckItem
{
public AckItem(string address, int handle)
{
Address = address;
Handle = handle;
}
public string Address { get; }
public int Handle { get; }
/// <summary>Set when the last write to this handle ended Unconfirmed; forces a fresh handle next time.</summary>
public bool Dirty { get; set; }
}
}
@@ -41,6 +41,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
private readonly IMxAccessComObjectFactory? comFactory;
private readonly Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory;
private readonly Func<IAlarmAckWriter>? ackWriterFactory;
private readonly Action? ackPumpStep;
private readonly MxAccessEventMapper mapper = new MxAccessEventMapper();
private readonly object syncRoot = new object();
private readonly IAlarmAckWriter deferredAckWriter;
@@ -131,13 +132,21 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
/// unsubscribe/resubscribe, and is disposed with the handler. Tests
/// inject a fake.
/// </param>
/// <param name="ackPumpStep">
/// Pump step the default <see cref="AlarmAckWriter"/> drives while it waits
/// for OnWriteComplete. Production passes the STA runtime's
/// <c>StaRuntime.PumpPendingMessages</c>, which also refreshes the STA
/// watchdog heartbeat; when <see langword="null"/> a bare
/// <see cref="StaMessagePump"/> is used (no heartbeat refresh).
/// </param>
public AlarmCommandHandler(
MxAccessEventQueue eventQueue,
Func<IMxAccessAlarmConsumer> consumerFactory,
Action? threadAffinityCheck,
IMxAccessComObjectFactory? comFactory,
Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory,
Func<IAlarmAckWriter>? ackWriterFactory = null)
Func<IAlarmAckWriter>? ackWriterFactory = null,
Action? ackPumpStep = null)
{
this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
this.consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
@@ -145,6 +154,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
this.comFactory = comFactory;
this.standbyFactory = standbyFactory;
this.ackWriterFactory = ackWriterFactory;
this.ackPumpStep = ackPumpStep;
this.deferredAckWriter = new DeferredAckWriter(this);
}
@@ -399,12 +409,14 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
"Alarm acknowledgement requires an IMxAccessComObjectFactory; the alarm command handler was constructed without one.");
}
// StaMessagePump.PumpPendingMessages is PeekMessage/DispatchMessage on
// the calling thread — the STA, since acks run there — which is what
// delivers the writer's OnWriteComplete callback.
// The pump step is PeekMessage/DispatchMessage on the calling thread —
// the STA, since acks run there — which is what delivers the writer's
// OnWriteComplete callback. The injected StaRuntime pump also marks STA
// activity so a long completion wait does not read as a hung STA.
Action pump = ackPumpStep ?? (() => new StaMessagePump().PumpPendingMessages());
ackWriter = new AlarmAckWriter(
comFactory,
() => new StaMessagePump().PumpPendingMessages(),
pump,
MxAccessStaSession.ResolveWriteCompletionTimeout(),
AlarmAckWriter.ResolveClientName());
return ackWriter;
@@ -48,7 +48,10 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
private readonly IMxAccessEventSink eventSink;
private readonly MxAccessEventQueue eventQueue;
private readonly StaRuntime staRuntime;
private readonly Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, IAlarmCommandHandler>? alarmCommandHandlerFactory;
// Pump-aware form: the fourth argument is the STA runtime's pump step, so
// handler work that pumps while it waits (the alarm ack writer's
// OnWriteComplete wait) also refreshes the STA watchdog heartbeat.
private readonly Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, Action, IAlarmCommandHandler>? alarmCommandHandlerFactory;
private StaCommandDispatcher? commandDispatcher;
private MxAccessSession? session;
private IAlarmCommandHandler? alarmCommandHandler;
@@ -96,6 +99,23 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
{
}
/// <summary>
/// Initializes a new instance of <see cref="MxAccessStaSession"/> with default dependencies and a
/// pump-aware alarm-command handler factory. Its fourth argument is a pump step bound to this
/// session's <see cref="StaRuntime"/> (<see cref="StaRuntime.PumpPendingMessages"/>), which
/// refreshes the STA watchdog heartbeat on every pump.
/// </summary>
/// <param name="pumpAwareAlarmCommandHandlerFactory">Factory that constructs the alarm-command handler.</param>
internal MxAccessStaSession(Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, Action, IAlarmCommandHandler> pumpAwareAlarmCommandHandlerFactory)
{
staRuntime = new StaRuntime();
factory = new MxAccessComObjectFactory();
eventQueue = new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity());
eventSink = new MxAccessBaseEventSink(eventQueue);
alarmCommandHandlerFactory = pumpAwareAlarmCommandHandlerFactory
?? throw new ArgumentNullException(nameof(pumpAwareAlarmCommandHandlerFactory));
}
/// <summary>
/// Initializes a new instance of <see cref="MxAccessStaSession"/> with custom STA runtime and factory.
/// </summary>
@@ -183,7 +203,9 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
this.alarmCommandHandlerFactory = alarmCommandHandlerFactory;
this.alarmCommandHandlerFactory = alarmCommandHandlerFactory is null
? null
: (eq, affinity, comFactory, _) => alarmCommandHandlerFactory(eq, affinity, comFactory);
}
/// <summary>
@@ -288,7 +310,8 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
alarmCommandHandler = alarmCommandHandlerFactory(
eventQueue,
EnsureOnAlarmConsumerThread,
factory);
factory,
() => staRuntime.PumpPendingMessages());
}
commandDispatcher = new StaCommandDispatcher(
staRuntime,
@@ -184,6 +184,57 @@ public sealed class MxAccessWriteCompletionCache
}
}
/// <summary>
/// Forgets the completion entry of a handle pair whose MXAccess item was
/// removed, so a long-lived writer that drops and re-adds items does not
/// keep one entry per dropped handle for the life of the worker.
/// </summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle that was removed.</param>
public void Remove(
int serverHandle,
int itemHandle)
{
lock (syncRoot)
{
entries.Remove(CreateItemKey(serverHandle, itemHandle));
}
}
/// <summary>Forgets every completion entry recorded under a server handle that was unregistered.</summary>
/// <param name="serverHandle">MXAccess server handle that was unregistered.</param>
public void RemoveServer(int serverHandle)
{
lock (syncRoot)
{
List<long> stale = new();
foreach (long key in entries.Keys)
{
if ((int)(key >> 32) == serverHandle)
{
stale.Add(key);
}
}
foreach (long key in stale)
{
entries.Remove(key);
}
}
}
/// <summary>Gets the number of handle pairs with a recorded completion (test seam).</summary>
internal int Count
{
get
{
lock (syncRoot)
{
return entries.Count;
}
}
}
private static long CreateItemKey(
int serverHandle,
int itemHandle)