diff --git a/clients/rust/crates/mxgw-cli/src/main.rs b/clients/rust/crates/mxgw-cli/src/main.rs index 075f0a3..0dace53 100644 --- a/clients/rust/crates/mxgw-cli/src/main.rs +++ b/clients/rust/crates/mxgw-cli/src/main.rs @@ -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"]); diff --git a/docs/Authorization.md b/docs/Authorization.md index 7d02679..c22425a 100644 --- a/docs/Authorization.md +++ b/docs/Authorization.md @@ -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: diff --git a/docs/DesignDecisions.md b/docs/DesignDecisions.md index 66ff30b..d8ed73d 100644 --- a/docs/DesignDecisions.md +++ b/docs/DesignDecisions.md @@ -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) diff --git a/docs/Grpc.md b/docs/Grpc.md index 93333bd..cdccebc 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -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, or the monitor's worker session having no active alarm subscription — `InvalidOperationException`) | `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. diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs index fc1630b..2332d8d 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs @@ -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,130 @@ 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_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_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne() { @@ -180,6 +308,10 @@ internal sealed class RecordingAckServer : IMxAccessServer public Action? OnWrite { get; set; } + public List RemovedItemHandles { get; } = new(); + + public List WrittenItemHandles { get; } = new(); + public int Register(string clientName) { Operations.Add("Register:" + clientName); @@ -196,7 +328,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 +352,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; diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index c9afa8d..f4f2de0 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -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 { diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs index ce9bb39..fafe755 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs @@ -26,13 +26,25 @@ public sealed class AlarmAckWriter : IAlarmAckWriter /// Default MXAccess client name when none is configured. public const string DefaultClientName = "MxGateway"; + /// + /// 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. + /// + 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 itemHandles = new Dictionary(StringComparer.OrdinalIgnoreCase); + private readonly int maxCachedItems; + private readonly Dictionary> itemHandles = new Dictionary>(StringComparer.OrdinalIgnoreCase); + + // Most-recently-used first; the tail is the eviction candidate. + private readonly LinkedList itemLru = new LinkedList(); private object? comObject; private LMXProxyServerClass? comEventSource; private IMxAccessServer? server; @@ -54,6 +66,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 +76,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)); @@ -95,8 +111,22 @@ public sealed class AlarmAckWriter : IAlarmAckWriter { userId = mx.AuthenticateUser(serverHandle, ackOperator.VerifyUser, ackOperator.Password ?? string.Empty); } - catch (COMException ex) + catch (Exception ex) when (IsInteropFailure(ex)) { + // Bad credentials surface as ArgumentException (E_INVALIDARG) from the + // interop, not only COMException, so classify by the carried HRESULT. + 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, @@ -113,14 +143,32 @@ public sealed class AlarmAckWriter : IAlarmAckWriter ulong baselineVersion = 0; int itemHandle = 0; - bool cached = itemHandles.TryGetValue(itemAddress, out itemHandle); + LinkedListNode? node; + bool cached = itemHandles.TryGetValue(itemAddress, out node); string step = "AddItem"; try { - if (!cached) + if (cached && node!.Value.Dirty) + { + // 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. + DropItem(mx, node); + cached = false; + } + + if (cached) + { + itemHandle = node!.Value.Handle; + itemLru.Remove(node); + itemLru.AddFirst(node); + } + else { itemHandle = mx.AddItem(serverHandle, itemAddress); - itemHandles[itemAddress] = itemHandle; + node = CacheItem(mx, itemAddress, itemHandle); step = "Advise"; // MXAccess rejects writes to added-but-unadvised items with E_INVALIDARG. mx.Advise(serverHandle, itemHandle); @@ -131,13 +179,15 @@ public sealed class AlarmAckWriter : IAlarmAckWriter baselineVersion = completionCache.CurrentVersion(serverHandle, itemHandle); mx.Write(serverHandle, itemHandle, comment ?? string.Empty, userId); } - catch (COMException ex) + catch (Exception ex) when (IsInteropFailure(ex)) { + // The interop maps E_INVALIDARG to ArgumentException (and + // E_ACCESSDENIED to UnauthorizedAccessException), so the HRESULT, not + // the exception type, is what identifies the MXAccess failure. int hresult = ex.HResult; - if (itemHandles.Remove(itemAddress)) + if (itemHandles.TryGetValue(itemAddress, out LinkedListNode? failed)) { - try { mx.UnAdvise(serverHandle, itemHandle); } catch { /* best effort */ } - try { mx.RemoveItem(serverHandle, itemHandle); } catch { /* best effort */ } + DropItem(mx, failed!); } if (IsServerGone(hresult)) @@ -162,6 +212,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter pumpStep, out RepeatedField statuses)) { + node!.Value.Dirty = true; return AlarmAckResult.Create( AlarmAckOutcome.Unconfirmed, 0, @@ -192,6 +243,17 @@ public sealed class AlarmAckWriter : IAlarmAckWriter TearDownServer(); } + /// + /// Exceptions the MXAccess COM interop raises for a failed HRESULT: the + /// runtime maps well-known HRESULTs to typed exceptions (E_INVALIDARG to + /// ArgumentException, E_ACCESSDENIED to UnauthorizedAccessException) and + /// the rest to COMException. Each carries the HRESULT in Exception.HResult. + /// + private static bool IsInteropFailure(Exception ex) => + ex is ExternalException + || ex is ArgumentException + || ex is UnauthorizedAccessException; + private static bool IsServerGone(int hresult) => hresult == unchecked((int)0x80010108) // RPC_E_DISCONNECTED || hresult == unchecked((int)0x800706BA) // RPC_S_SERVER_UNAVAILABLE @@ -208,16 +270,17 @@ 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 */ } } itemHandles.Clear(); + itemLru.Clear(); server = null; comEventSource = null; object? toRelease = comObject; @@ -228,6 +291,28 @@ public sealed class AlarmAckWriter : IAlarmAckWriter } } + /// Caches a newly added item as most recently used, evicting the least recently used past the cap. + private LinkedListNode CacheItem(IMxAccessServer mx, string itemAddress, int itemHandle) + { + while (itemLru.Count >= maxCachedItems && itemLru.Last is { } leastRecent) + { + DropItem(mx, leastRecent); + } + + LinkedListNode node = itemLru.AddFirst(new AckItem(itemAddress, itemHandle)); + itemHandles[itemAddress] = node; + return node; + } + + /// Removes a cached item and best-effort releases its MXAccess item. + private void DropItem(IMxAccessServer mx, LinkedListNode node) + { + itemHandles.Remove(node.Value.Address); + itemLru.Remove(node); + try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ } + try { mx.RemoveItem(serverHandle, node.Value.Handle); } catch { /* best effort */ } + } + private IMxAccessServer EnsureRegistered() { if (server is not null) return server; @@ -276,4 +361,21 @@ public sealed class AlarmAckWriter : IAlarmAckWriter rows.AddRange(statusConverter.ConvertMany(pVars)); completionCache.Record(hLMXServerHandle, phItemHandle, rows); } + + /// A cached ack item. Plain class (not a record) for the net48 worker target. + private sealed class AckItem + { + public AckItem(string address, int handle) + { + Address = address; + Handle = handle; + } + + public string Address { get; } + + public int Handle { get; } + + /// Set when the last write to this handle ended Unconfirmed; forces a fresh handle next time. + public bool Dirty { get; set; } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs index 3648511..67273e9 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs @@ -41,6 +41,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler private readonly IMxAccessComObjectFactory? comFactory; private readonly Func, IMxAccessAlarmConsumer>? standbyFactory; private readonly Func? 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. /// + /// + /// Pump step the default drives while it waits + /// for OnWriteComplete. Production passes the STA runtime's + /// StaRuntime.PumpPendingMessages, which also refreshes the STA + /// watchdog heartbeat; when a bare + /// is used (no heartbeat refresh). + /// public AlarmCommandHandler( MxAccessEventQueue eventQueue, Func consumerFactory, Action? threadAffinityCheck, IMxAccessComObjectFactory? comFactory, Func, IMxAccessAlarmConsumer>? standbyFactory, - Func? ackWriterFactory = null) + Func? 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; diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index ef9022a..086188f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -48,7 +48,10 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession private readonly IMxAccessEventSink eventSink; private readonly MxAccessEventQueue eventQueue; private readonly StaRuntime staRuntime; - private readonly Func? 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? alarmCommandHandlerFactory; private StaCommandDispatcher? commandDispatcher; private MxAccessSession? session; private IAlarmCommandHandler? alarmCommandHandler; @@ -96,6 +99,23 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession { } + /// + /// Initializes a new instance of with default dependencies and a + /// pump-aware alarm-command handler factory. Its fourth argument is a pump step bound to this + /// session's (), which + /// refreshes the STA watchdog heartbeat on every pump. + /// + /// Factory that constructs the alarm-command handler. + internal MxAccessStaSession(Func pumpAwareAlarmCommandHandlerFactory) + { + staRuntime = new StaRuntime(); + factory = new MxAccessComObjectFactory(); + eventQueue = new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity()); + eventSink = new MxAccessBaseEventSink(eventQueue); + alarmCommandHandlerFactory = pumpAwareAlarmCommandHandlerFactory + ?? throw new ArgumentNullException(nameof(pumpAwareAlarmCommandHandlerFactory)); + } + /// /// Initializes a new instance of with custom STA runtime and factory. /// @@ -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); } /// @@ -288,7 +310,8 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession alarmCommandHandler = alarmCommandHandlerFactory( eventQueue, EnsureOnAlarmConsumerThread, - factory); + factory, + () => staRuntime.PumpPendingMessages()); } commandDispatcher = new StaCommandDispatcher( staRuntime,