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);
},