getStatusesList() {
@@ -53991,6 +54180,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
}
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public int getStatusesCount() {
@@ -54001,6 +54205,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
}
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy getStatuses(int index) {
@@ -54011,6 +54230,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
}
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder setStatuses(
@@ -54028,6 +54262,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder setStatuses(
@@ -54042,6 +54291,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder addStatuses(mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy value) {
@@ -54058,6 +54322,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder addStatuses(
@@ -54075,6 +54354,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder addStatuses(
@@ -54089,6 +54383,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder addStatuses(
@@ -54103,6 +54412,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder addAllStatuses(
@@ -54118,6 +54442,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder clearStatuses() {
@@ -54131,6 +54470,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public Builder removeStatuses(int index) {
@@ -54144,6 +54498,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this;
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder getStatusesBuilder(
@@ -54151,6 +54520,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return internalGetStatusesFieldBuilder().getBuilder(index);
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder getStatusesOrBuilder(
@@ -54161,6 +54545,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
}
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public java.util.List
@@ -54172,6 +54571,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
}
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder addStatusesBuilder() {
@@ -54179,6 +54593,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.getDefaultInstance());
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder addStatusesBuilder(
@@ -54187,6 +54616,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
index, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.getDefaultInstance());
}
/**
+ *
+ * Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ * replies the worker holds the reply for a bounded window (default 1.5 s,
+ * MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ * MXAccess OnWriteComplete callback and copies its status rows here, so
+ * statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ * while protocol_status/hresult still describe command acceptance only.
+ * Empty statuses on a write reply means the completion did not arrive
+ * within the window — the write is unconfirmed, not failed. Correlation is
+ * best-effort per (server_handle, item_handle): MXAccess's callback carries
+ * no transaction id, so concurrent writes to the same item within the
+ * window can swap rows. The OnWriteComplete event still flows on the event
+ * stream unchanged. Other command kinds leave this field as before.
+ *
+ *
* repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;
*/
public java.util.List
diff --git a/clients/rust/protos/mxaccess_gateway.proto b/clients/rust/protos/mxaccess_gateway.proto
index be96d5a..ee923b5 100644
--- a/clients/rust/protos/mxaccess_gateway.proto
+++ b/clients/rust/protos/mxaccess_gateway.proto
@@ -256,6 +256,9 @@ message Write2Command {
int32 user_id = 5;
}
+// The unary reply's statuses field carries the correlated OnWriteComplete
+// outcome when it arrives within the worker's bounded wait — see
+// MxCommandReply.statuses.
message WriteSecuredCommand {
int32 server_handle = 1;
int32 item_handle = 2;
@@ -266,6 +269,9 @@ message WriteSecuredCommand {
MxValue value = 5;
}
+// The unary reply's statuses field carries the correlated OnWriteComplete
+// outcome when it arrives within the worker's bounded wait — see
+// MxCommandReply.statuses.
message WriteSecured2Command {
int32 server_handle = 1;
int32 item_handle = 2;
@@ -525,6 +531,18 @@ message MxCommandReply {
// transport failures.
optional int32 hresult = 5;
MxValue return_value = 6;
+ // Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ // replies the worker holds the reply for a bounded window (default 1.5 s,
+ // MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ // MXAccess OnWriteComplete callback and copies its status rows here, so
+ // statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ // while protocol_status/hresult still describe command acceptance only.
+ // Empty statuses on a write reply means the completion did not arrive
+ // within the window — the write is unconfirmed, not failed. Correlation is
+ // best-effort per (server_handle, item_handle): MXAccess's callback carries
+ // no transaction id, so concurrent writes to the same item within the
+ // window can swap rows. The OnWriteComplete event still flows on the event
+ // stream unchanged. Other command kinds leave this field as before.
repeated MxStatusProxy statuses = 7;
string diagnostic_message = 8;
diff --git a/docs/DesignDecisions.md b/docs/DesignDecisions.md
index ecb6169..6b14485 100644
--- a/docs/DesignDecisions.md
+++ b/docs/DesignDecisions.md
@@ -534,6 +534,48 @@ against the live MXAccess attribute set.
- [Alarm Client Discovery — Subtag provider](./AlarmClientDiscovery.md)
- [gRPC Contract — provider_status and degraded fields](./Grpc.md)
+## Secured-Write Completion Correlation
+
+MXAccess writes are fire-and-forget: the toolkit call returns before the
+Galaxy commit, and the per-item outcome only exists in the later
+`OnWriteComplete` COM callback. The original unary write reply therefore
+proved worker-side command acceptance only, forcing consumers (OtOpcUa's
+GalaxyDriver) to report every write as provisionally good.
+
+For `WriteSecured`/`WriteSecured2` the worker now holds the unary reply for a
+bounded window (`MxGateway:Worker:WriteCompletionWaitMilliseconds`, default
+1.5 s, `0` disables; conveyed to the worker via
+`MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS`) and copies the matching
+callback's status rows onto `MxCommandReply.statuses`. Key choices, argued in
+[the design doc](./plans/2026-08-09-write-completion-correlation-design.md):
+
+- **Pump-wait on the STA, not a parked reply.** The executor holds the STA
+ thread but pumps Windows messages each poll — the shipped ReadBulk pattern —
+ because commands serialize per session anyway, so freeing the STA during the
+ wait buys nothing and a parked reply would change the dispatcher/pipe
+ contracts.
+- **Version baseline before the COM call** closes the fast-completion edge: a
+ callback that dispatches while `WriteSecured` is still on the stack still
+ correlates.
+- **Timeout returns today's shape** (protocol OK, empty statuses):
+ unconfirmed is honest; a synthesized failure row would trigger consumer-side
+ write-revert logic on slow-but-successful commits. The 1.5 s default stays
+ inside OtOpcUa's 2 s Tier A write-resilience budget.
+- **Parity preserved.** `protocol_status`/`hresult` keep describing
+ acceptance; the MXAccess outcome (success or failure) rides only in
+ `statuses[0]`; the `OnWriteComplete` event still streams unchanged (nothing
+ swallowed, nothing synthesized).
+- **Scope: secured writes only.** Plain `Write`/`Write2` and bulk writes stay
+ fire-and-forget — the wait would add a device round-trip per write to
+ high-rate supervisory loops.
+- **Best-effort correlation.** The callback carries only
+ `(hItem, statuses)` — no transaction id — so concurrent writes to the same
+ item within the window can swap rows; benign for the serialized single-write
+ consumer contract.
+- **Client cancellation needs no special path**: a caller abandoning the RPC
+ mid-wait leaves the worker to finish its bounded wait and reply; the gateway
+ discards the reply, the session is never faulted.
+
## Later Revisit Items
These are explicit post-v1 revisit items, not open blockers:
diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md
index f77e08f..58ca7a3 100644
--- a/docs/GatewayConfiguration.md
+++ b/docs/GatewayConfiguration.md
@@ -114,6 +114,7 @@ launch CWD (SEC-01, SEC-33).
| `MxGateway:Worker:StartupProbeRetryAttempts` | `3` | Number of retry attempts for transient worker startup probe failures before pipe connection and handshake continue. |
| `MxGateway:Worker:StartupProbeRetryDelayMilliseconds` | `250` | Delay between transient startup probe retry attempts. |
| `MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds` | `2000` | Per-attempt timeout used by the worker named-pipe connect retry path. The overall pipe connection still stays under the startup budget. |
+| `MxGateway:Worker:WriteCompletionWaitMilliseconds` | `1500` | Bounded wait the worker holds a `WriteSecured`/`WriteSecured2` reply for the matching MXAccess `OnWriteComplete` callback, so the reply's `statuses` carry the real commit outcome. `0` disables the wait (pure fire-and-forget replies). Must be `>= 0`. The gateway conveys the value to the worker via the `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` environment variable. Consumers that time their own writes must budget above this wait: OtOpcUa's GalaxyDriver wraps gateway writes in a 2 s Tier A resilience timeout, so a deployment raising this option past ~2000 must raise that driver `ResilienceConfig` write timeout in step or slow-but-successful commits surface as consumer-side failures. |
| `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. |
| `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. |
| `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. |
diff --git a/gateway.md b/gateway.md
index 460f120..0f68da5 100644
--- a/gateway.md
+++ b/gateway.md
@@ -431,6 +431,23 @@ Core commands:
- `AuthenticateUser`
- `ArchestrAUserToId`
+**Secured-write completion correlation.** MXAccess writes are fire-and-forget
+at the toolkit level — the per-item outcome only exists in the later
+`OnWriteComplete` callback. For `WriteSecured` and `WriteSecured2` the worker
+therefore holds the unary reply for a bounded window
+(`MxGateway:Worker:WriteCompletionWaitMilliseconds`, default 1.5 s, `0`
+disables) and, when the matching callback arrives, copies its status rows onto
+`MxCommandReply.statuses` — the reply then proves the MXAccess commit, not just
+command acceptance. `protocol_status`/`hresult` keep describing acceptance
+only; a real MXAccess write failure surfaces in `statuses[0]`, and a reply with
+empty `statuses` means unconfirmed (the callback missed the window), never
+failed. The `OnWriteComplete` event still flows on the event stream unchanged.
+Correlation is best-effort per `(server_handle, item_handle)` — the callback
+carries no transaction id, so concurrent writes to the same item within the
+window can swap rows. Plain `Write`/`Write2` and the bulk write commands stay
+fire-and-forget: waiting there would add a device round-trip of latency to
+high-rate supervisory write loops.
+
Bulk variants (single gRPC round-trip carries the full list, the worker
runs the per-item MXAccess calls sequentially on its STA, and the reply
returns one result per requested entry — per-entry failures populate
diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs
index f78dc5b..400e221 100644
--- a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs
+++ b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessGateway.cs
@@ -8694,6 +8694,11 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
}
+ ///
+ /// The unary reply's statuses field carries the correlated OnWriteComplete
+ /// outcome when it arrives within the worker's bounded wait — see
+ /// MxCommandReply.statuses.
+ ///
[global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
public sealed partial class WriteSecuredCommand : pb::IMessage
#if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
@@ -9053,6 +9058,11 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
}
+ ///
+ /// The unary reply's statuses field carries the correlated OnWriteComplete
+ /// outcome when it arrives within the worker's bounded wait — see
+ /// MxCommandReply.statuses.
+ ///
[global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
public sealed partial class WriteSecured2Command : pb::IMessage
#if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
@@ -17204,6 +17214,20 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
private static readonly pb::FieldCodec _repeated_statuses_codec
= pb::FieldCodec.ForMessage(58, global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy.Parser);
private readonly pbc::RepeatedField statuses_ = new pbc::RepeatedField();
+ ///
+ /// Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ /// replies the worker holds the reply for a bounded window (default 1.5 s,
+ /// MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ /// MXAccess OnWriteComplete callback and copies its status rows here, so
+ /// statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ /// while protocol_status/hresult still describe command acceptance only.
+ /// Empty statuses on a write reply means the completion did not arrive
+ /// within the window — the write is unconfirmed, not failed. Correlation is
+ /// best-effort per (server_handle, item_handle): MXAccess's callback carries
+ /// no transaction id, so concurrent writes to the same item within the
+ /// window can swap rows. The OnWriteComplete event still flows on the event
+ /// stream unchanged. Other command kinds leave this field as before.
+ ///
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public pbc::RepeatedField Statuses {
diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto
index be96d5a..ee923b5 100644
--- a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto
+++ b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto
@@ -256,6 +256,9 @@ message Write2Command {
int32 user_id = 5;
}
+// The unary reply's statuses field carries the correlated OnWriteComplete
+// outcome when it arrives within the worker's bounded wait — see
+// MxCommandReply.statuses.
message WriteSecuredCommand {
int32 server_handle = 1;
int32 item_handle = 2;
@@ -266,6 +269,9 @@ message WriteSecuredCommand {
MxValue value = 5;
}
+// The unary reply's statuses field carries the correlated OnWriteComplete
+// outcome when it arrives within the worker's bounded wait — see
+// MxCommandReply.statuses.
message WriteSecured2Command {
int32 server_handle = 1;
int32 item_handle = 2;
@@ -525,6 +531,18 @@ message MxCommandReply {
// transport failures.
optional int32 hresult = 5;
MxValue return_value = 6;
+ // Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
+ // replies the worker holds the reply for a bounded window (default 1.5 s,
+ // MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
+ // MXAccess OnWriteComplete callback and copies its status rows here, so
+ // statuses[0] carries the real MXAccess commit outcome (success OR failure)
+ // while protocol_status/hresult still describe command acceptance only.
+ // Empty statuses on a write reply means the completion did not arrive
+ // within the window — the write is unconfirmed, not failed. Correlation is
+ // best-effort per (server_handle, item_handle): MXAccess's callback carries
+ // no transaction id, so concurrent writes to the same item within the
+ // window can swap rows. The OnWriteComplete event still flows on the event
+ // stream unchanged. Other command kinds leave this field as before.
repeated MxStatusProxy statuses = 7;
string diagnostic_message = 8;
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs
index 05a6967..2d18295 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs
@@ -225,6 +225,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBaseThe timeout in milliseconds for connecting to the worker pipe.
public int PipeConnectAttemptTimeoutMilliseconds { get; init; } = 2000;
+ ///
+ /// Bounded wait, in milliseconds, the worker holds a WriteSecured/WriteSecured2
+ /// reply for the matching MXAccess OnWriteComplete callback so the reply's
+ /// statuses carry the real commit outcome. 0 disables the wait. Deployments
+ /// raising this above consumer write-timeout budgets (e.g. OtOpcUa's 2 s Tier A
+ /// write resilience timeout) must raise those in step.
+ ///
+ public int WriteCompletionWaitMilliseconds { get; init; } = 1500;
+
/// The maximum time in seconds for graceful shutdown.
public int ShutdownTimeoutSeconds { get; init; } = 10;
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs
index cab6fe1..b74f85a 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs
@@ -21,6 +21,14 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName =
"MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS";
+ ///
+ /// Conveys MxGateway:Worker:WriteCompletionWaitMilliseconds to the worker:
+ /// the bounded wait for the OnWriteComplete callback on
+ /// WriteSecured/WriteSecured2 replies. 0 disables the wait.
+ ///
+ public const string WorkerWriteCompletionWaitEnvironmentVariableName =
+ "MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
+
private readonly IWorkerProcessFactory _processFactory;
private readonly IWorkerStartupProbe _startupProbe;
private readonly GatewayMetrics _metrics;
@@ -175,6 +183,8 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
startInfo.Environment[WorkerNonceEnvironmentVariableName] = request.Nonce;
startInfo.Environment[WorkerPipeConnectAttemptTimeoutEnvironmentVariableName] =
_workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
+ startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
+ _workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
commandLine = new WorkerProcessCommandLine(executablePath, arguments);
diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs
index e05e93b..1506edc 100644
--- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs
@@ -41,6 +41,7 @@ public sealed class GatewayOptionsTests
Assert.Equal(3, options.Worker.StartupProbeRetryAttempts);
Assert.Equal(250, options.Worker.StartupProbeRetryDelayMilliseconds);
Assert.Equal(2000, options.Worker.PipeConnectAttemptTimeoutMilliseconds);
+ Assert.Equal(1500, options.Worker.WriteCompletionWaitMilliseconds);
Assert.Equal(10, options.Worker.ShutdownTimeoutSeconds);
Assert.Equal(5, options.Worker.HeartbeatIntervalSeconds);
Assert.Equal(15, options.Worker.HeartbeatGraceSeconds);
@@ -106,6 +107,7 @@ public sealed class GatewayOptionsTests
[InlineData("MxGateway:Worker:ExecutablePath", "worker.dll", "MxGateway:Worker:ExecutablePath must point to a .exe file.")]
[InlineData("MxGateway:Worker:StartupProbeRetryAttempts", "0", "MxGateway:Worker:StartupProbeRetryAttempts must be greater than zero.")]
[InlineData("MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds", "0", "MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds must be greater than zero.")]
+ [InlineData("MxGateway:Worker:WriteCompletionWaitMilliseconds", "-1", "MxGateway:Worker:WriteCompletionWaitMilliseconds must be greater than or equal to zero.")]
[InlineData("MxGateway:Sessions:DefaultLeaseSeconds", "0", "MxGateway:Sessions:DefaultLeaseSeconds must be greater than zero.")]
[InlineData("MxGateway:Sessions:LeaseSweepIntervalSeconds", "0", "MxGateway:Sessions:LeaseSweepIntervalSeconds must be greater than zero.")]
[InlineData("MxGateway:Sessions:DetachGraceSeconds", "-1", "MxGateway:Sessions:DetachGraceSeconds must be zero or greater (0 disables detach-grace retention).")]
diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs
index f9d0e43..52b3756 100644
--- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs
@@ -43,6 +43,10 @@ public sealed class WorkerProcessLauncherTests
"2000",
processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerPipeConnectAttemptTimeoutEnvironmentVariableName]);
+ Assert.Equal(
+ "1500",
+ processFactory.LastStartInfo.Environment[
+ WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]);
Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal);
Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal);
Assert.False(pipeReservation.DisposeCalled);
diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs
index 935e9d2..a81362e 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs
@@ -110,6 +110,47 @@ public sealed class MxAccessBaseEventSinkTests
Assert.Same(cache, sink.ValueCache);
}
+ ///
+ /// Verifies that an OnWriteComplete COM callback records the completion into
+ /// the per-session write-completion cache for the executor's bounded reply
+ /// wait AND still enqueues the event unchanged for the outbound stream —
+ /// correlation observes the event, it never consumes it. The cache update
+ /// fires only after the event has cleared the queue (post-publish rule).
+ ///
+ [Fact]
+ public void OnWriteComplete_ComCallback_RecordsCompletionAndStillEnqueuesEvent()
+ {
+ MxAccessEventQueue queue = new();
+ MxAccessWriteCompletionCache completionCache = new();
+ MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper(), new MxAccessValueCache(), completionCache);
+ MXSTATUS_PROXY[] statuses = Array.Empty();
+
+ sink.OnWriteComplete(hLMXServerHandle: 7, phItemHandle: 21, ref statuses);
+
+ Assert.Equal(1, queue.Count);
+ Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent));
+ MxEvent mxEvent = workerEvent!.Event;
+ Assert.Equal(MxEventFamily.OnWriteComplete, mxEvent.Family);
+ Assert.Equal(7, mxEvent.ServerHandle);
+ Assert.Equal(21, mxEvent.ItemHandle);
+ Assert.Equal(1UL, completionCache.CurrentVersion(7, 21));
+ }
+
+ ///
+ /// Verifies that the sink-bound write-completion cache is exposed for sharing
+ /// with the owning so the sink's recordings and
+ /// the write executor's waits see the same instance.
+ ///
+ [Fact]
+ public void WriteCompletionCache_ReturnsTheInstanceBoundAtConstruction()
+ {
+ MxAccessEventQueue queue = new();
+ MxAccessWriteCompletionCache completionCache = new();
+ MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper(), new MxAccessValueCache(), completionCache);
+
+ Assert.Same(completionCache, sink.WriteCompletionCache);
+ }
+
///
/// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences.
///
diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs
index 2b4b0aa..34f9406 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs
@@ -897,6 +897,9 @@ public sealed class MxAccessCommandExecutorTests
FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+ // No completion source in this test — disable the bounded reply wait
+ // so the forwarding assertions don't pay the default 1.5 s timeout.
+ session.WriteCompletionTimeout = TimeSpan.Zero;
await session.StartAsync(workerProcessId: 1234);
MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
@@ -919,6 +922,8 @@ public sealed class MxAccessCommandExecutorTests
FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+ // Same rationale as the WriteSecured forwarding test above.
+ session.WriteCompletionTimeout = TimeSpan.Zero;
await session.StartAsync(workerProcessId: 1234);
DateTime timestamp = new(2026, 5, 19, 13, 30, 0, DateTimeKind.Utc);
@@ -934,6 +939,192 @@ public sealed class MxAccessCommandExecutorTests
Assert.Equal(44, fakeComObject.WriteVerifierUserId);
}
+ ///
+ /// Verifies the fast-completion ordering edge: a completion recorded while
+ /// the WriteSecured COM call is still on the stack (MXAccess committing
+ /// synchronously) is newer than the pre-call baseline and lands on the
+ /// reply — the wait never misses a callback that beat it.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_WriteSecured_WhenCompletionArrivesDuringComCall_ReturnsStatuses()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 82);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ fakeComObject.OnWriteSecuredCallback = () =>
+ sink.WriteCompletionCache.Record(82, 820, CreateCompletionRows(detail: 4321));
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ // Hermetic: don't inherit MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS
+ // from the test runner's environment.
+ session.WriteCompletionTimeout = TimeSpan.FromSeconds(10);
+ await session.StartAsync(workerProcessId: 1234);
+
+ MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
+ "write-secured-fast", serverHandle: 82, itemHandle: 820, value: 1, currentUserId: 11, verifierUserId: 22));
+
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.True(reply.HasHresult);
+ Assert.Equal(0, reply.Hresult);
+ MxStatusProxy row = Assert.Single(reply.Statuses);
+ Assert.Equal(4321, row.Detail);
+ Assert.Equal(MxStatusCategory.Ok, row.Category);
+ }
+
+ ///
+ /// Verifies the pump-wait path: the completion arrives after the COM call
+ /// returned, while the executor is pump-waiting, and still lands on the
+ /// reply.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_WriteSecured_WhenCompletionArrivesWhileWaiting_ReturnsStatuses()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 83);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ // Deterministic ordering: the executor captures its version baseline
+ // BEFORE the COM call, so once the fake's WriteSecured has run the
+ // baseline is committed and a Record from the test thread is
+ // guaranteed to be "newer" — no fixed sleep racing the STA thread.
+ using System.Threading.ManualResetEventSlim comCallReached = new(initialState: false);
+ fakeComObject.OnWriteSecuredCallback = () => comCallReached.Set();
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ session.WriteCompletionTimeout = TimeSpan.FromSeconds(10);
+ await session.StartAsync(workerProcessId: 1234);
+
+ Task pending = session.DispatchAsync(CreateWriteSecuredCommand(
+ "write-secured-waiting", serverHandle: 83, itemHandle: 830, value: 1, currentUserId: 11, verifierUserId: 22));
+ Assert.True(comCallReached.Wait(TimeSpan.FromSeconds(5)));
+ sink.WriteCompletionCache.Record(83, 830, CreateCompletionRows(detail: 99));
+
+ MxCommandReply reply = await pending;
+
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.Equal(99, Assert.Single(reply.Statuses).Detail);
+ }
+
+ ///
+ /// Verifies the timeout fallback: no completion within the bounded wait
+ /// returns today's reply shape — protocol OK with EMPTY statuses (the
+ /// consumer's honest-unconfirmed path), never a synthesized failure row.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_WriteSecured_WhenNoCompletion_TimesOutWithEmptyStatusesAndOkProtocol()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 84);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ session.WriteCompletionTimeout = TimeSpan.FromMilliseconds(100);
+ await session.StartAsync(workerProcessId: 1234);
+
+ MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
+ "write-secured-timeout", serverHandle: 84, itemHandle: 840, value: 1, currentUserId: 11, verifierUserId: 22));
+
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.True(reply.HasHresult);
+ Assert.Equal(0, reply.Hresult);
+ Assert.Empty(reply.Statuses);
+ }
+
+ ///
+ /// Verifies the version-baseline rule end to end: a completion recorded
+ /// BEFORE the write was dispatched is stale and must not be misattributed
+ /// to this write — the reply times out empty instead.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_WriteSecured_IgnoresStaleCompletionFromBeforeTheCall()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 85);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ sink.WriteCompletionCache.Record(85, 850, CreateCompletionRows(detail: 1111));
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ session.WriteCompletionTimeout = TimeSpan.FromMilliseconds(100);
+ await session.StartAsync(workerProcessId: 1234);
+
+ MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
+ "write-secured-stale", serverHandle: 85, itemHandle: 850, value: 1, currentUserId: 11, verifierUserId: 22));
+
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.Empty(reply.Statuses);
+ }
+
+ ///
+ /// Verifies that WriteSecured2 correlates the same way as WriteSecured
+ /// (fast-completion edge).
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_WriteSecured2_WhenCompletionArrivesDuringComCall_ReturnsStatuses()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 86);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ fakeComObject.OnWriteSecuredCallback = () =>
+ sink.WriteCompletionCache.Record(86, 860, CreateCompletionRows(detail: 2222));
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ // Hermetic: same rationale as the WriteSecured fast-completion test.
+ session.WriteCompletionTimeout = TimeSpan.FromSeconds(10);
+ await session.StartAsync(workerProcessId: 1234);
+
+ MxCommandReply reply = await session.DispatchAsync(CreateWriteSecured2Command(
+ "write-secured2-fast", serverHandle: 86, itemHandle: 860, value: 1,
+ timestamp: new DateTime(2026, 8, 9, 12, 0, 0, DateTimeKind.Utc), currentUserId: 33, verifierUserId: 44));
+
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.Equal(2222, Assert.Single(reply.Statuses).Detail);
+ }
+
+ ///
+ /// Verifies plain Write stays fire-and-forget: even with a huge completion
+ /// timeout configured and no completion source, the reply returns
+ /// immediately (guarded well under the configured wait) with empty
+ /// statuses — only the secured write kinds enter the bounded wait.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task DispatchAsync_Write_DoesNotWaitForCompletion()
+ {
+ FakeMxAccessComObject fakeComObject = new(registerHandle: 87);
+ FakeMxAccessComObjectFactory factory = new(fakeComObject);
+ CompletionCacheEventSink sink = new();
+ using StaRuntime runtime = CreateRuntime();
+ using MxAccessStaSession session = new(runtime, factory, sink);
+ session.WriteCompletionTimeout = TimeSpan.FromSeconds(30);
+ await session.StartAsync(workerProcessId: 1234);
+
+ Task pending = session.DispatchAsync(CreateWriteCommand(
+ "plain-write-no-wait", serverHandle: 87, itemHandle: 870, value: 1, userId: 5));
+ Task completed = await Task.WhenAny(pending, Task.Delay(TimeSpan.FromSeconds(5)));
+
+ Assert.Same(pending, completed);
+ MxCommandReply reply = await pending;
+ Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+ Assert.Empty(reply.Statuses);
+ }
+
+ private static Google.Protobuf.Collections.RepeatedField CreateCompletionRows(int detail)
+ {
+ return new Google.Protobuf.Collections.RepeatedField
+ {
+ new MxStatusProxy
+ {
+ Success = 1,
+ Category = MxStatusCategory.Ok,
+ Detail = detail,
+ },
+ };
+ }
+
/// Verifies that Write without a payload returns an invalid request error.
/// A task that represents the asynchronous operation.
[Fact]
@@ -1775,6 +1966,28 @@ public sealed class MxAccessCommandExecutorTests
TimeSpan.FromMilliseconds(25));
}
+ ///
+ /// Test sink that owns a real write-completion cache without touching the
+ /// MXAccess COM RCW (Attach is a no-op). Implements the provider seam so
+ /// shares this cache with the write
+ /// executor, letting tests record completions the executor's bounded
+ /// wait then observes.
+ ///
+ private sealed class CompletionCacheEventSink : IMxAccessEventSink, IWriteCompletionCacheProvider
+ {
+ public MxAccessWriteCompletionCache WriteCompletionCache { get; } = new MxAccessWriteCompletionCache();
+
+ public void Attach(
+ object mxAccessComObject,
+ string sessionId)
+ {
+ }
+
+ public void Detach()
+ {
+ }
+ }
+
private sealed class FakeMxAccessComObject : IMxAccessServer
{
private readonly int registerHandle;
@@ -1790,6 +2003,14 @@ public sealed class MxAccessCommandExecutorTests
private readonly IReadOnlyDictionary writeExceptionByItemHandle;
private readonly List operationNames = new();
+ ///
+ /// Invoked at the end of a successful WriteSecured/WriteSecured2 —
+ /// stands in for MXAccess committing synchronously and delivering
+ /// OnWriteComplete while the COM call is still on the stack, so
+ /// tests can exercise the fast-completion ordering edge.
+ ///
+ public Action? OnWriteSecuredCallback { get; set; }
+
/// Initializes a fake MXAccess COM object with the given handles and optional exceptions.
/// Return value for Register method.
/// Return value for AddItem method.
@@ -2100,6 +2321,7 @@ public sealed class MxAccessCommandExecutorTests
WriteValue = value;
WriteThreadId = Environment.CurrentManagedThreadId;
ThrowIfWriteFailureConfigured(itemHandle);
+ OnWriteSecuredCallback?.Invoke();
}
///
@@ -2120,6 +2342,7 @@ public sealed class MxAccessCommandExecutorTests
WriteTimestamp = timestamp;
WriteThreadId = Environment.CurrentManagedThreadId;
ThrowIfWriteFailureConfigured(itemHandle);
+ OnWriteSecuredCallback?.Invoke();
}
private void ThrowIfWriteFailureConfigured(int itemHandle)
diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs
index a8002b3..44894ea 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs
@@ -15,6 +15,48 @@ namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
///
public sealed class MxAccessStaSessionTests
{
+ ///
+ /// Verifies the launcher-env-var parse branches of
+ /// :
+ /// a valid non-negative value is honored (0 = disabled), while a
+ /// missing, malformed, or negative value falls back to the executor
+ /// default. Env mutation is restored in a finally so parallel tests
+ /// never observe the temporary value.
+ ///
+ /// Raw env-var value, or null for unset.
+ /// Expected resolved wait, or null for the executor default.
+ [Theory]
+ [InlineData(null, null)]
+ [InlineData("", null)]
+ [InlineData("junk", null)]
+ [InlineData("-5", null)]
+ [InlineData("0", 0)]
+ [InlineData("250", 250)]
+ public void ResolveWriteCompletionTimeout_ParsesEnvironmentValue(string? rawValue, int? expectedMilliseconds)
+ {
+ string? original = Environment.GetEnvironmentVariable(
+ MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName);
+ try
+ {
+ Environment.SetEnvironmentVariable(
+ MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName,
+ rawValue);
+
+ TimeSpan resolved = MxAccessStaSession.ResolveWriteCompletionTimeout();
+
+ TimeSpan expected = expectedMilliseconds is null
+ ? MxAccessCommandExecutor.DefaultWriteCompletionTimeout
+ : TimeSpan.FromMilliseconds(expectedMilliseconds.Value);
+ Assert.Equal(expected, resolved);
+ }
+ finally
+ {
+ Environment.SetEnvironmentVariable(
+ MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName,
+ original);
+ }
+ }
+
///
/// Verifies that StartAsync creates the MXAccess COM object and attaches the event sink on the STA thread.
///
diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs
new file mode 100644
index 0000000..670bc5f
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs
@@ -0,0 +1,145 @@
+using System;
+using Google.Protobuf.Collections;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
+
+///
+/// Unit tests for . The cache is
+/// consumed by the write command executor's bounded pump-wait so a
+/// WriteSecured/WriteSecured2 reply can carry the correlated
+/// OnWriteComplete outcome; its version-baseline contract is exercised in
+/// isolation here before the STA / COM plumbing gets layered on top.
+///
+public sealed class MxAccessWriteCompletionCacheTests
+{
+ /// Verifies that Record bumps the version per key and keys stay isolated.
+ [Fact]
+ public void Record_IncrementsVersionPerKey()
+ {
+ MxAccessWriteCompletionCache cache = new();
+
+ Assert.Equal(0UL, cache.CurrentVersion(7, 21));
+
+ cache.Record(7, 21, BuildStatuses(detail: 100));
+ Assert.Equal(1UL, cache.CurrentVersion(7, 21));
+
+ cache.Record(7, 21, BuildStatuses(detail: 200));
+ Assert.Equal(2UL, cache.CurrentVersion(7, 21));
+
+ cache.Record(7, 22, BuildStatuses(detail: 300));
+ Assert.Equal(1UL, cache.CurrentVersion(7, 22));
+ Assert.Equal(2UL, cache.CurrentVersion(7, 21));
+ }
+
+ /// Verifies that a completion newer than the baseline is returned with its status rows.
+ [Fact]
+ public void TryWaitForCompletion_WhenCompletionNewerThanBaseline_ReturnsStatuses()
+ {
+ MxAccessWriteCompletionCache cache = new();
+ cache.Record(7, 21, BuildStatuses(detail: 4321));
+
+ bool found = cache.TryWaitForCompletion(
+ 7,
+ 21,
+ sinceVersion: 0UL,
+ deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+ pumpStep: static () => { },
+ out RepeatedField statuses);
+
+ Assert.True(found);
+ MxStatusProxy row = Assert.Single(statuses);
+ Assert.Equal(4321, row.Detail);
+ Assert.Equal(MxStatusCategory.Ok, row.Category);
+ }
+
+ ///
+ /// Verifies that a completion recorded before the baseline was captured is
+ /// never misattributed to the waiting write: only a strictly newer version
+ /// satisfies the wait, so a stale row times the wait out.
+ ///
+ [Fact]
+ public void TryWaitForCompletion_WhenOnlyStaleCompletion_TimesOut()
+ {
+ MxAccessWriteCompletionCache cache = new();
+ cache.Record(7, 21, BuildStatuses(detail: 4321));
+ ulong baseline = cache.CurrentVersion(7, 21);
+
+ bool found = cache.TryWaitForCompletion(
+ 7,
+ 21,
+ sinceVersion: baseline,
+ deadlineUtc: DateTime.UtcNow.AddMilliseconds(50),
+ pumpStep: static () => { },
+ out RepeatedField statuses);
+
+ Assert.False(found);
+ Assert.Empty(statuses);
+ }
+
+ ///
+ /// Verifies the pump loop is what lets a completion land: the completion is
+ /// recorded from inside a later pump step (standing in for the STA
+ /// dispatching the OnWriteComplete message) and the wait then succeeds.
+ ///
+ [Fact]
+ public void TryWaitForCompletion_InvokesPumpStepEachIteration()
+ {
+ MxAccessWriteCompletionCache cache = new();
+ int pumpCalls = 0;
+
+ bool found = cache.TryWaitForCompletion(
+ 7,
+ 21,
+ sinceVersion: 0UL,
+ deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+ pumpStep: () =>
+ {
+ pumpCalls++;
+ if (pumpCalls == 2)
+ {
+ cache.Record(7, 21, BuildStatuses(detail: 55));
+ }
+ },
+ out RepeatedField statuses);
+
+ Assert.True(found);
+ Assert.True(pumpCalls >= 2);
+ Assert.Equal(55, Assert.Single(statuses).Detail);
+ }
+
+ /// Verifies that Record stores an independent clone of the caller's rows.
+ [Fact]
+ public void Record_ClonesStatuses()
+ {
+ MxAccessWriteCompletionCache cache = new();
+ RepeatedField callerRows = BuildStatuses(detail: 77);
+
+ cache.Record(7, 21, callerRows);
+ callerRows[0].Detail = 999;
+ callerRows.Add(new MxStatusProxy());
+
+ Assert.True(cache.TryWaitForCompletion(
+ 7,
+ 21,
+ sinceVersion: 0UL,
+ deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+ pumpStep: static () => { },
+ out RepeatedField statuses));
+ Assert.Equal(77, Assert.Single(statuses).Detail);
+ }
+
+ private static RepeatedField BuildStatuses(int detail)
+ {
+ return new RepeatedField
+ {
+ new MxStatusProxy
+ {
+ Success = 1,
+ Category = MxStatusCategory.Ok,
+ Detail = detail,
+ },
+ };
+ }
+}
diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs
new file mode 100644
index 0000000..71509e9
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs
@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+///
+/// Exposes the per-session an
+/// event sink populates from OnWriteComplete callbacks, so
+/// can share one instance between the
+/// sink (writer) and the write command executor (reader). Implemented by
+/// and by test sinks that cannot
+/// attach to a live MXAccess COM object.
+///
+public interface IWriteCompletionCacheProvider
+{
+ /// The completion cache bound to this sink.
+ MxAccessWriteCompletionCache WriteCompletionCache { get; }
+}
diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs
index 33c0bfe..d3903c9 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs
@@ -5,11 +5,12 @@ using Proto = ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// Sink for MXAccess COM events that converts them to protobuf format.
-public sealed class MxAccessBaseEventSink : IMxAccessEventSink
+public sealed class MxAccessBaseEventSink : IMxAccessEventSink, IWriteCompletionCacheProvider
{
private readonly MxAccessEventMapper eventMapper;
private readonly MxAccessEventQueue eventQueue;
private readonly MxAccessValueCache valueCache;
+ private readonly MxAccessWriteCompletionCache writeCompletionCache;
private LMXProxyServerClass? server;
private string sessionId = string.Empty;
@@ -50,10 +51,32 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
MxAccessEventQueue eventQueue,
MxAccessEventMapper eventMapper,
MxAccessValueCache valueCache)
+ : this(eventQueue, eventMapper, valueCache, new MxAccessWriteCompletionCache())
+ {
+ }
+
+ ///
+ /// Initializes a new instance of the MxAccessBaseEventSink class with
+ /// provided queue, mapper, value cache, and a shared write-completion
+ /// cache. The completion cache is populated from every successful
+ /// OnWriteComplete dispatch so the worker's write executor can
+ /// correlate a WriteSecured/WriteSecured2 reply with the MXAccess
+ /// completion outcome.
+ ///
+ /// Queue for buffering converted MXAccess events.
+ /// Converter for MXAccess events to protobuf format.
+ /// Per-session last-value cache shared with the MxAccessSession.
+ /// Per-session OnWriteComplete cache shared with the MxAccessSession.
+ public MxAccessBaseEventSink(
+ MxAccessEventQueue eventQueue,
+ MxAccessEventMapper eventMapper,
+ MxAccessValueCache valueCache,
+ MxAccessWriteCompletionCache writeCompletionCache)
{
this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
+ this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
}
///
@@ -62,6 +85,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
///
public MxAccessValueCache ValueCache => valueCache;
+ ///
+ /// The OnWriteComplete completion cache populated by this sink. Exposed
+ /// via so the
+ /// MxAccessSession can share the same instance with the write command
+ /// executor's bounded completion wait.
+ ///
+ public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;
+
///
public void Attach(
object mxAccessComObject,
@@ -143,11 +174,18 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
ref MXSTATUS_PROXY[] pVars)
{
MXSTATUS_PROXY[] statuses = pVars;
- EnqueueEvent(() => eventMapper.CreateOnWriteComplete(
- sessionId,
- hLMXServerHandle,
- phItemHandle,
- statuses));
+ // Record the completion for the write executor's bounded reply wait
+ // only after the event has cleared the queue (same post-publish rule
+ // as the OnDataChange value cache) — an overflow faults the session,
+ // so a dropped event never leaves a "fresher" completion behind than
+ // what shipped to the gateway.
+ EnqueueEvent(
+ () => eventMapper.CreateOnWriteComplete(
+ sessionId,
+ hLMXServerHandle,
+ phItemHandle,
+ statuses),
+ mxEvent => writeCompletionCache.Record(hLMXServerHandle, phItemHandle, mxEvent.Statuses));
}
///
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/MxAccessSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs
index a82e0a6..78f1c0d 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs
@@ -13,6 +13,7 @@ public sealed class MxAccessSession : IDisposable
private readonly IMxAccessEventSink eventSink;
private readonly MxAccessHandleRegistry handleRegistry;
private readonly MxAccessValueCache valueCache;
+ private readonly MxAccessWriteCompletionCache writeCompletionCache;
private bool disposed;
private MxAccessSession(
@@ -21,6 +22,7 @@ public sealed class MxAccessSession : IDisposable
IMxAccessEventSink eventSink,
MxAccessHandleRegistry handleRegistry,
MxAccessValueCache valueCache,
+ MxAccessWriteCompletionCache writeCompletionCache,
int creationThreadId)
{
this.mxAccessComObject = mxAccessComObject ?? throw new ArgumentNullException(nameof(mxAccessComObject));
@@ -28,6 +30,7 @@ public sealed class MxAccessSession : IDisposable
this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
this.handleRegistry = handleRegistry ?? throw new ArgumentNullException(nameof(handleRegistry));
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
+ this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
CreationThreadId = creationThreadId;
}
@@ -45,6 +48,14 @@ public sealed class MxAccessSession : IDisposable
///
public MxAccessValueCache ValueCache => valueCache;
+ ///
+ /// Per-session OnWriteComplete completion cache populated by the event
+ /// sink. The write command executor consults it after a
+ /// WriteSecured/WriteSecured2 COM call so the unary reply can carry
+ /// the correlated completion outcome.
+ ///
+ public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;
+
/// Creates a WorkerReady message with session metadata.
/// Process ID of the worker.
/// The populated message.
@@ -105,6 +116,9 @@ public sealed class MxAccessSession : IDisposable
eventSink,
handleRegistry ?? new MxAccessHandleRegistry(),
valueCache ?? new MxAccessValueCache(),
+ eventSink is IWriteCompletionCacheProvider provider
+ ? provider.WriteCompletionCache
+ : new MxAccessWriteCompletionCache(),
creationThreadId ?? Environment.CurrentManagedThreadId);
}
@@ -149,12 +163,22 @@ public sealed class MxAccessSession : IDisposable
? baseSink.ValueCache
: new MxAccessValueCache();
+ // Share the sink's completion cache the same way (the production
+ // sink and completion-aware test sinks implement the provider
+ // seam); fall back to a fresh cache for other fakes — the write
+ // executor then simply never observes a completion and replies
+ // unconfirmed.
+ MxAccessWriteCompletionCache writeCompletionCache = eventSink is IWriteCompletionCacheProvider provider
+ ? provider.WriteCompletionCache
+ : new MxAccessWriteCompletionCache();
+
return new MxAccessSession(
mxAccessComObject,
new MxAccessComServer(mxAccessComObject),
eventSink,
new MxAccessHandleRegistry(),
valueCache,
+ writeCompletionCache,
Environment.CurrentManagedThreadId);
}
catch (Exception exception)
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);
},
diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs
new file mode 100644
index 0000000..1bb5e35
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs
@@ -0,0 +1,152 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using Google.Protobuf.Collections;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+///
+/// Per-session cache of the most recent OnWriteComplete status rows
+/// for each (server handle, item handle) pair. Written by the MXAccess
+/// event sink as completion callbacks arrive; read by the write command
+/// executor so a WriteSecured/WriteSecured2 reply can carry the correlated
+/// MXAccess outcome instead of proving command acceptance only.
+///
+///
+/// Same threading posture as : writers and
+/// readers run on the worker's STA thread (COM dispatches events on the
+/// apartment thread; commands also execute on the STA), so no internal
+/// locking is required. A single sync root keeps it nominally thread-safe
+/// for tests that drive it from a non-STA thread.
+///
+public sealed class MxAccessWriteCompletionCache
+{
+ private readonly Dictionary entries = new();
+ private readonly object syncRoot = new();
+
+ /// Records the status rows of a fresh OnWriteComplete callback for the given handle pair.
+ /// MXAccess server handle.
+ /// MXAccess item handle.
+ /// Status rows from the mapped OnWriteComplete event; cloned before storing.
+ public void Record(
+ int serverHandle,
+ int itemHandle,
+ RepeatedField statuses)
+ {
+ if (statuses is null)
+ {
+ throw new ArgumentNullException(nameof(statuses));
+ }
+
+ lock (syncRoot)
+ {
+ long key = CreateItemKey(serverHandle, itemHandle);
+ ulong version = entries.TryGetValue(key, out CompletionEntry existing)
+ ? existing.Version + 1
+ : 1UL;
+ entries[key] = new CompletionEntry(version, statuses.Clone());
+ }
+ }
+
+ /// Returns the current completion version for a handle pair, or 0 if none was recorded.
+ /// MXAccess server handle.
+ /// MXAccess item handle.
+ /// The current completion version, or 0 if no completion was recorded.
+ public ulong CurrentVersion(
+ int serverHandle,
+ int itemHandle)
+ {
+ lock (syncRoot)
+ {
+ return entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry existing)
+ ? existing.Version
+ : 0UL;
+ }
+ }
+
+ ///
+ /// Polls for a completion newer than until it
+ /// arrives or the deadline elapses, calling on every
+ /// poll iteration so the worker's STA can dispatch the inbound MXAccess
+ /// OnWriteComplete message. Same loop shape as
+ /// .
+ ///
+ /// MXAccess server handle.
+ /// MXAccess item handle.
+ /// Version snapshot captured before the write COM call.
+ /// Absolute UTC deadline.
+ /// Action that pumps any pending Windows messages.
+ /// The recorded status rows if a completion arrived before the deadline; empty otherwise.
+ /// How long to sleep between pump cycles. Default 5 ms.
+ /// if a completion newer than arrived before the deadline; otherwise .
+ public bool TryWaitForCompletion(
+ int serverHandle,
+ int itemHandle,
+ ulong sinceVersion,
+ DateTime deadlineUtc,
+ Action pumpStep,
+ out RepeatedField statuses,
+ int pollIntervalMs = 5)
+ {
+ if (pumpStep is null)
+ {
+ throw new ArgumentNullException(nameof(pumpStep));
+ }
+
+ while (true)
+ {
+ pumpStep();
+
+ lock (syncRoot)
+ {
+ if (entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry entry)
+ && entry.Version > sinceVersion)
+ {
+ statuses = entry.Statuses;
+ return true;
+ }
+ }
+
+ if (DateTime.UtcNow >= deadlineUtc)
+ {
+ statuses = new RepeatedField();
+ return false;
+ }
+
+ Thread.Sleep(pollIntervalMs);
+ }
+ }
+
+ private static long CreateItemKey(
+ int serverHandle,
+ int itemHandle)
+ {
+ return ((long)serverHandle << 32) | (uint)itemHandle;
+ }
+
+ ///
+ /// Snapshot of the most recent OnWriteComplete status rows for a handle
+ /// pair. increments by one on every
+ /// call so the write executor can detect "a new
+ /// completion arrived since I captured my baseline".
+ ///
+ ///
+ /// Plain readonly struct (not a record) so this compiles under the
+ /// worker's net48 target, which lacks IsExternalInit.
+ ///
+ private readonly struct CompletionEntry
+ {
+ public CompletionEntry(
+ ulong version,
+ RepeatedField statuses)
+ {
+ Version = version;
+ Statuses = statuses;
+ }
+
+ public ulong Version { get; }
+
+ public RepeatedField Statuses { get; }
+ }
+}