feat(worker): bounded pump-wait correlates OnWriteComplete onto secured-write replies

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs

@@ -14,11 +14,22 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
     /// <summary>Default per-tag timeout used when <c>ReadBulkCommand.timeout_ms</c> is zero.</summary>
     internal static readonly TimeSpan DefaultReadBulkTimeout = TimeSpan.FromMilliseconds(1000);
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    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;
 
     /// <summary>
     /// Initializes a command executor with an MXAccess session.
@@ -71,17 +82,26 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
     /// <param name="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
     /// <param name="alarmCommandHandler">Optional handler for alarm-side commands.</param>
     /// <param name="pumpStep">Action to pump Windows messages, or null for tests.</param>
+    /// <param name="writeCompletionTimeout">
+    ///     Bounded wait for the OnWriteComplete callback after a
+    ///     WriteSecured/WriteSecured2 COM call, or null for
+    ///     <see cref="DefaultWriteCompletionTimeout"/>. Zero (or negative)
+    ///     disables the wait entirely — replies keep the pure fire-and-forget
+    ///     shape.
+    /// </param>
     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;
     }
 
     /// <inheritdoc />
@@ -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
         }
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    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<MxStatusProxy> statuses))
+        {
+            reply.Statuses.Add(statuses);
+        }
+    }
+
     private static MxCommandReply CreateAlarmFailureReply(StaCommand command, Exception exception)
     {
         return new MxCommandReply

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
 {
+    /// <summary>
+    ///     Environment variable the gateway's WorkerProcessLauncher sets from
+    ///     MxGateway:Worker:WriteCompletionWaitMilliseconds. 0 disables the
+    ///     write-completion wait (pure fire-and-forget replies).
+    /// </summary>
+    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
     /// </summary>
     public MxAccessEventQueue EventQueue => eventQueue;
 
+    /// <summary>
+    ///     Bounded WriteSecured/WriteSecured2 completion wait handed to the
+    ///     command executor at <see cref="StartAsync(string, int, CancellationToken)"/>.
+    ///     Internal-settable as a test seam so Worker.Tests can shorten it
+    ///     without env-var plumbing.
+    /// </summary>
+    internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout();
+
+    /// <summary>
+    ///     Resolves the write-completion wait from the launcher-provided
+    ///     environment variable; a missing or invalid value falls back to
+    ///     <see cref="MxAccessCommandExecutor.DefaultWriteCompletionTimeout"/>.
+    /// </summary>
+    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;
+    }
+
     /// <summary>
     /// Starts the MXAccess COM session asynchronously.
     /// </summary>
@@ -208,12 +242,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
                         session,
                         new VariantConverter(),
                         alarmCommandHandler,
-                        // ReadBulk needs to pump Windows messages while it waits
+                        // ReadBulk and the write-completion wait need to pump
-                        // for the first OnDataChange callback so the inbound COM
+                        // Windows messages while they wait for the inbound COM
-                        // event can dispatch on this same STA thread. The pump
+                        // callback (OnDataChange / OnWriteComplete) so it can
-                        // step closes over staRuntime so it always pumps the
+                        // dispatch on this same STA thread. The pump step
-                        // pump tied to the apartment that owns this session.
+                        // closes over staRuntime so it always pumps the pump
-                        pumpStep: () => staRuntime.PumpPendingMessages()));
+                        // tied to the apartment that owns this session.
+                        pumpStep: () => staRuntime.PumpPendingMessages(),
+                        writeCompletionTimeout: WriteCompletionTimeout));
 
                 return session.CreateWorkerReady(workerProcessId);
             },