From 66fe063410adafd36a45a18130066e9b6e7c83ab Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 9 Aug 2026 12:24:23 -0400 Subject: [PATCH] feat(worker): bounded pump-wait correlates OnWriteComplete onto secured-write replies --- .../MxAccess/MxAccessCommandExecutor.cs | 87 ++++++++++++++++++- .../MxAccess/MxAccessStaSession.cs | 48 ++++++++-- 2 files changed, 126 insertions(+), 9 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs index da91f4b..6b061bd 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs @@ -14,11 +14,22 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor /// Default per-tag timeout used when ReadBulkCommand.timeout_ms is zero. internal static readonly TimeSpan DefaultReadBulkTimeout = TimeSpan.FromMilliseconds(1000); + /// + /// Default bounded wait for the OnWriteComplete callback after a + /// WriteSecured/WriteSecured2 COM call. 1.5 s keeps the unary reply + /// inside the OtOpcUa driver's 2 s Tier A write-resilience budget (a + /// longer gateway wait must raise that consumer timeout in step) while + /// covering the common fast-commit case; on expiry the reply returns + /// with empty statuses — unconfirmed, not failed. + /// + internal static readonly TimeSpan DefaultWriteCompletionTimeout = TimeSpan.FromMilliseconds(1500); + private readonly MxAccessSession session; private readonly VariantConverter variantConverter; private readonly MxStatusProxyConverter statusProxyConverter; private readonly IAlarmCommandHandler? alarmCommandHandler; private readonly Action pumpStep; + private readonly TimeSpan writeCompletionTimeout; /// /// Initializes a command executor with an MXAccess session. @@ -71,17 +82,26 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor /// Converter for MXAccess variant values to MxValue protobuf messages. /// Optional handler for alarm-side commands. /// Action to pump Windows messages, or null for tests. + /// + /// Bounded wait for the OnWriteComplete callback after a + /// WriteSecured/WriteSecured2 COM call, or null for + /// . Zero (or negative) + /// disables the wait entirely — replies keep the pure fire-and-forget + /// shape. + /// public MxAccessCommandExecutor( MxAccessSession session, VariantConverter variantConverter, IAlarmCommandHandler? alarmCommandHandler, - Action? pumpStep) + Action? pumpStep, + TimeSpan? writeCompletionTimeout = null) { this.session = session ?? throw new ArgumentNullException(nameof(session)); this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter)); this.statusProxyConverter = new MxStatusProxyConverter(); this.alarmCommandHandler = alarmCommandHandler; this.pumpStep = pumpStep ?? (static () => { }); + this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout; } /// @@ -457,6 +477,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor return CreateInvalidRequestReply(command, "WriteSecured command value is required."); } + // Baseline BEFORE the COM call: a completion that dispatches during or + // immediately after WriteSecured bumps the version past this snapshot, + // so a fast commit still correlates (no missed-callback window). + MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache; + ulong completionBaseline = completionCache.CurrentVersion( + writeSecuredCommand.ServerHandle, + writeSecuredCommand.ItemHandle); + session.WriteSecured( writeSecuredCommand.ServerHandle, writeSecuredCommand.ItemHandle, @@ -464,7 +492,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor writeSecuredCommand.VerifierUserId, variantConverter.ConvertToComValue(writeSecuredCommand.Value)); - return CreateOkReply(command); + MxCommandReply reply = CreateOkReply(command); + AwaitWriteCompletion( + reply, + completionCache, + writeSecuredCommand.ServerHandle, + writeSecuredCommand.ItemHandle, + completionBaseline); + return reply; } private MxCommandReply ExecuteWriteSecured2(StaCommand command) @@ -485,6 +520,12 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor return CreateInvalidRequestReply(command, "WriteSecured2 command timestamp value is required."); } + // Same pre-call baseline rule as ExecuteWriteSecured. + MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache; + ulong completionBaseline = completionCache.CurrentVersion( + writeSecured2Command.ServerHandle, + writeSecured2Command.ItemHandle); + session.WriteSecured2( writeSecured2Command.ServerHandle, writeSecured2Command.ItemHandle, @@ -493,7 +534,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor variantConverter.ConvertToComValue(writeSecured2Command.Value), variantConverter.ConvertToComValue(writeSecured2Command.TimestampValue)); - return CreateOkReply(command); + MxCommandReply reply = CreateOkReply(command); + AwaitWriteCompletion( + reply, + completionCache, + writeSecured2Command.ServerHandle, + writeSecured2Command.ItemHandle, + completionBaseline); + return reply; } private MxCommandReply ExecuteAddItemBulk(StaCommand command) @@ -897,6 +945,39 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor } } + /// + /// Bounded pump-wait for the OnWriteComplete row matching a + /// WriteSecured/WriteSecured2 call, copied onto the reply when it + /// arrives in time. The executor holds the STA thread but pumps + /// Windows messages each poll (ReadBulk precedent) so the COM callback + /// can dispatch re-entrantly; on expiry the reply keeps its empty + /// statuses — the consumer's unconfirmed path, never a synthesized + /// failure. Protocol status/hresult stay acceptance-only either way. + /// + private void AwaitWriteCompletion( + MxCommandReply reply, + MxAccessWriteCompletionCache completionCache, + int serverHandle, + int itemHandle, + ulong completionBaseline) + { + if (writeCompletionTimeout <= TimeSpan.Zero) + { + return; + } + + if (completionCache.TryWaitForCompletion( + serverHandle, + itemHandle, + completionBaseline, + DateTime.UtcNow + writeCompletionTimeout, + pumpStep, + out Google.Protobuf.Collections.RepeatedField statuses)) + { + reply.Statuses.Add(statuses); + } + } + private static MxCommandReply CreateAlarmFailureReply(StaCommand command, Exception exception) { return new MxCommandReply diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 51ccbab..90e622f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -11,6 +11,14 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; public sealed class MxAccessStaSession : IWorkerRuntimeSession { + /// + /// Environment variable the gateway's WorkerProcessLauncher sets from + /// MxGateway:Worker:WriteCompletionWaitMilliseconds. 0 disables the + /// write-completion wait (pure fire-and-forget replies). + /// + internal const string WriteCompletionWaitEnvironmentVariableName = + "MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS"; + private static readonly TimeSpan AlarmPollInterval = TimeSpan.FromMilliseconds(500); private readonly IMxAccessComObjectFactory factory; @@ -157,6 +165,32 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession /// public MxAccessEventQueue EventQueue => eventQueue; + /// + /// Bounded WriteSecured/WriteSecured2 completion wait handed to the + /// command executor at . + /// Internal-settable as a test seam so Worker.Tests can shorten it + /// without env-var plumbing. + /// + internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout(); + + /// + /// Resolves the write-completion wait from the launcher-provided + /// environment variable; a missing or invalid value falls back to + /// . + /// + internal static TimeSpan ResolveWriteCompletionTimeout() + { + string? value = Environment.GetEnvironmentVariable(WriteCompletionWaitEnvironmentVariableName); + return int.TryParse( + value, + System.Globalization.NumberStyles.Integer, + System.Globalization.CultureInfo.InvariantCulture, + out int milliseconds) + && milliseconds >= 0 + ? TimeSpan.FromMilliseconds(milliseconds) + : MxAccessCommandExecutor.DefaultWriteCompletionTimeout; + } + /// /// Starts the MXAccess COM session asynchronously. /// @@ -208,12 +242,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession session, new VariantConverter(), alarmCommandHandler, - // ReadBulk needs to pump Windows messages while it waits - // for the first OnDataChange callback so the inbound COM - // event can dispatch on this same STA thread. The pump - // step closes over staRuntime so it always pumps the - // pump tied to the apartment that owns this session. - pumpStep: () => staRuntime.PumpPendingMessages())); + // ReadBulk and the write-completion wait need to pump + // Windows messages while they wait for the inbound COM + // callback (OnDataChange / OnWriteComplete) so it can + // dispatch on this same STA thread. The pump step + // closes over staRuntime so it always pumps the pump + // tied to the apartment that owns this session. + pumpStep: () => staRuntime.PumpPendingMessages(), + writeCompletionTimeout: WriteCompletionTimeout)); return session.CreateWorkerReady(workerProcessId); },