feat: correlate OnWriteComplete onto WriteSecured/WriteSecured2 unary replies (OtOpcUa 06/S-1)
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m9s
ci / java (push) Successful in 2m3s
ci / portable (push) Successful in 7m20s

This commit is contained in:
Joseph Doherty
2026-08-09 12:52:18 -04:00
23 changed files with 1425 additions and 17 deletions
@@ -2697,6 +2697,9 @@ func (x *Write2Command) GetUserId() int32 {
return 0 return 0
} }
// The unary reply's statuses field carries the correlated OnWriteComplete
// outcome when it arrives within the worker's bounded wait — see
// MxCommandReply.statuses.
type WriteSecuredCommand struct { type WriteSecuredCommand struct {
state protoimpl.MessageState `protogen:"open.v1"` state protoimpl.MessageState `protogen:"open.v1"`
ServerHandle int32 `protobuf:"varint,1,opt,name=server_handle,json=serverHandle,proto3" json:"server_handle,omitempty"` ServerHandle int32 `protobuf:"varint,1,opt,name=server_handle,json=serverHandle,proto3" json:"server_handle,omitempty"`
@@ -2775,6 +2778,9 @@ func (x *WriteSecuredCommand) GetValue() *MxValue {
return nil return nil
} }
// The unary reply's statuses field carries the correlated OnWriteComplete
// outcome when it arrives within the worker's bounded wait — see
// MxCommandReply.statuses.
type WriteSecured2Command struct { type WriteSecured2Command struct {
state protoimpl.MessageState `protogen:"open.v1"` state protoimpl.MessageState `protogen:"open.v1"`
ServerHandle int32 `protobuf:"varint,1,opt,name=server_handle,json=serverHandle,proto3" json:"server_handle,omitempty"` ServerHandle int32 `protobuf:"varint,1,opt,name=server_handle,json=serverHandle,proto3" json:"server_handle,omitempty"`
@@ -4577,6 +4583,18 @@ type MxCommandReply struct {
// transport failures. // transport failures.
Hresult *int32 `protobuf:"varint,5,opt,name=hresult,proto3,oneof" json:"hresult,omitempty"` Hresult *int32 `protobuf:"varint,5,opt,name=hresult,proto3,oneof" json:"hresult,omitempty"`
ReturnValue *MxValue `protobuf:"bytes,6,opt,name=return_value,json=returnValue,proto3" json:"return_value,omitempty"` ReturnValue *MxValue `protobuf:"bytes,6,opt,name=return_value,json=returnValue,proto3" json:"return_value,omitempty"`
// 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.
Statuses []*MxStatusProxy `protobuf:"bytes,7,rep,name=statuses,proto3" json:"statuses,omitempty"` Statuses []*MxStatusProxy `protobuf:"bytes,7,rep,name=statuses,proto3" json:"statuses,omitempty"`
DiagnosticMessage string `protobuf:"bytes,8,opt,name=diagnostic_message,json=diagnosticMessage,proto3" json:"diagnostic_message,omitempty"` DiagnosticMessage string `protobuf:"bytes,8,opt,name=diagnostic_message,json=diagnosticMessage,proto3" json:"diagnostic_message,omitempty"`
// Types that are valid to be assigned to Payload: // Types that are valid to be assigned to Payload:
@@ -26028,6 +26028,12 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getValueOrBuilder(); mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getValueOrBuilder();
} }
/** /**
* <pre>
* The unary reply's statuses field carries the correlated OnWriteComplete
* outcome when it arrives within the worker's bounded wait see
* MxCommandReply.statuses.
* </pre>
*
* Protobuf type {@code mxaccess_gateway.v1.WriteSecuredCommand} * Protobuf type {@code mxaccess_gateway.v1.WriteSecuredCommand}
*/ */
public static final class WriteSecuredCommand extends public static final class WriteSecuredCommand extends
@@ -26357,6 +26363,12 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return builder; return builder;
} }
/** /**
* <pre>
* The unary reply's statuses field carries the correlated OnWriteComplete
* outcome when it arrives within the worker's bounded wait see
* MxCommandReply.statuses.
* </pre>
*
* Protobuf type {@code mxaccess_gateway.v1.WriteSecuredCommand} * Protobuf type {@code mxaccess_gateway.v1.WriteSecuredCommand}
*/ */
public static final class Builder extends public static final class Builder extends
@@ -26976,6 +26988,12 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getTimestampValueOrBuilder(); mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getTimestampValueOrBuilder();
} }
/** /**
* <pre>
* The unary reply's statuses field carries the correlated OnWriteComplete
* outcome when it arrives within the worker's bounded wait see
* MxCommandReply.statuses.
* </pre>
*
* Protobuf type {@code mxaccess_gateway.v1.WriteSecured2Command} * Protobuf type {@code mxaccess_gateway.v1.WriteSecured2Command}
*/ */
public static final class WriteSecured2Command extends public static final class WriteSecured2Command extends
@@ -27347,6 +27365,12 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return builder; return builder;
} }
/** /**
* <pre>
* The unary reply's statuses field carries the correlated OnWriteComplete
* outcome when it arrives within the worker's bounded wait see
* MxCommandReply.statuses.
* </pre>
*
* Protobuf type {@code mxaccess_gateway.v1.WriteSecured2Command} * Protobuf type {@code mxaccess_gateway.v1.WriteSecured2Command}
*/ */
public static final class Builder extends public static final class Builder extends
@@ -50488,24 +50512,99 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getReturnValueOrBuilder(); mxaccess_gateway.v1.MxaccessGateway.MxValueOrBuilder getReturnValueOrBuilder();
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy> java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy>
getStatusesList(); getStatusesList();
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy getStatuses(int index); mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy getStatuses(int index);
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
int getStatusesCount(); int getStatusesCount();
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
java.util.List<? extends mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder> java.util.List<? extends mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder>
getStatusesOrBuilderList(); getStatusesOrBuilderList();
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder getStatusesOrBuilder( mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder getStatusesOrBuilder(
@@ -51233,6 +51332,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
@SuppressWarnings("serial") @SuppressWarnings("serial")
private java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy> statuses_; private java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy> statuses_;
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
@java.lang.Override @java.lang.Override
@@ -51240,6 +51354,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return statuses_; return statuses_;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
@java.lang.Override @java.lang.Override
@@ -51248,6 +51377,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return statuses_; return statuses_;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
@java.lang.Override @java.lang.Override
@@ -51255,6 +51399,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return statuses_.size(); return statuses_.size();
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
@java.lang.Override @java.lang.Override
@@ -51262,6 +51421,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return statuses_.get(index); return statuses_.get(index);
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
@java.lang.Override @java.lang.Override
@@ -53981,6 +54155,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder> statusesBuilder_; mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder> statusesBuilder_;
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy> getStatusesList() { public java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy> getStatusesList() {
@@ -53991,6 +54180,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
} }
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public int getStatusesCount() { public int getStatusesCount() {
@@ -54001,6 +54205,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
} }
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy getStatuses(int index) { public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy getStatuses(int index) {
@@ -54011,6 +54230,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
} }
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder setStatuses( public Builder setStatuses(
@@ -54028,6 +54262,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder setStatuses( public Builder setStatuses(
@@ -54042,6 +54291,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder addStatuses(mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy value) { public Builder addStatuses(mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy value) {
@@ -54058,6 +54322,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder addStatuses( public Builder addStatuses(
@@ -54075,6 +54354,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder addStatuses( public Builder addStatuses(
@@ -54089,6 +54383,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder addStatuses( public Builder addStatuses(
@@ -54103,6 +54412,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder addAllStatuses( public Builder addAllStatuses(
@@ -54118,6 +54442,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder clearStatuses() { public Builder clearStatuses() {
@@ -54131,6 +54470,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public Builder removeStatuses(int index) { public Builder removeStatuses(int index) {
@@ -54144,6 +54498,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
return this; return this;
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder getStatusesBuilder( 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); return internalGetStatusesFieldBuilder().getBuilder(index);
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder getStatusesOrBuilder( public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder getStatusesOrBuilder(
@@ -54161,6 +54545,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
} }
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public java.util.List<? extends mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder> public java.util.List<? extends mxaccess_gateway.v1.MxaccessGateway.MxStatusProxyOrBuilder>
@@ -54172,6 +54571,21 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
} }
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder addStatusesBuilder() { 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()); mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.getDefaultInstance());
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder addStatusesBuilder( 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()); index, mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.getDefaultInstance());
} }
/** /**
* <pre>
* 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.
* </pre>
*
* <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code> * <code>repeated .mxaccess_gateway.v1.MxStatusProxy statuses = 7;</code>
*/ */
public java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder> public java.util.List<mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.Builder>
@@ -256,6 +256,9 @@ message Write2Command {
int32 user_id = 5; 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 { message WriteSecuredCommand {
int32 server_handle = 1; int32 server_handle = 1;
int32 item_handle = 2; int32 item_handle = 2;
@@ -266,6 +269,9 @@ message WriteSecuredCommand {
MxValue value = 5; 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 { message WriteSecured2Command {
int32 server_handle = 1; int32 server_handle = 1;
int32 item_handle = 2; int32 item_handle = 2;
@@ -525,6 +531,18 @@ message MxCommandReply {
// transport failures. // transport failures.
optional int32 hresult = 5; optional int32 hresult = 5;
MxValue return_value = 6; 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; repeated MxStatusProxy statuses = 7;
string diagnostic_message = 8; string diagnostic_message = 8;
+42
View File
@@ -534,6 +534,48 @@ against the live MXAccess attribute set.
- [Alarm Client Discovery — Subtag provider](./AlarmClientDiscovery.md) - [Alarm Client Discovery — Subtag provider](./AlarmClientDiscovery.md)
- [gRPC Contract — provider_status and degraded fields](./Grpc.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 ## Later Revisit Items
These are explicit post-v1 revisit items, not open blockers: These are explicit post-v1 revisit items, not open blockers:
+1
View File
@@ -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: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: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: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: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: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`. | | `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`. |
+17
View File
@@ -431,6 +431,23 @@ Core commands:
- `AuthenticateUser` - `AuthenticateUser`
- `ArchestrAUserToId` - `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 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 runs the per-item MXAccess calls sequentially on its STA, and the reply
returns one result per requested entry — per-entry failures populate returns one result per requested entry — per-entry failures populate
@@ -8694,6 +8694,11 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
} }
/// <summary>
/// The unary reply's statuses field carries the correlated OnWriteComplete
/// outcome when it arrives within the worker's bounded wait — see
/// MxCommandReply.statuses.
/// </summary>
[global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
public sealed partial class WriteSecuredCommand : pb::IMessage<WriteSecuredCommand> public sealed partial class WriteSecuredCommand : pb::IMessage<WriteSecuredCommand>
#if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
@@ -9053,6 +9058,11 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
} }
/// <summary>
/// The unary reply's statuses field carries the correlated OnWriteComplete
/// outcome when it arrives within the worker's bounded wait — see
/// MxCommandReply.statuses.
/// </summary>
[global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
public sealed partial class WriteSecured2Command : pb::IMessage<WriteSecured2Command> public sealed partial class WriteSecured2Command : pb::IMessage<WriteSecured2Command>
#if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
@@ -17204,6 +17214,20 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
private static readonly pb::FieldCodec<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> _repeated_statuses_codec private static readonly pb::FieldCodec<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> _repeated_statuses_codec
= pb::FieldCodec.ForMessage(58, global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy.Parser); = pb::FieldCodec.ForMessage(58, global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy.Parser);
private readonly pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> statuses_ = new pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy>(); private readonly pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> statuses_ = new pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy>();
/// <summary>
/// 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.
/// </summary>
[global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> Statuses { public pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> Statuses {
@@ -256,6 +256,9 @@ message Write2Command {
int32 user_id = 5; 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 { message WriteSecuredCommand {
int32 server_handle = 1; int32 server_handle = 1;
int32 item_handle = 2; int32 item_handle = 2;
@@ -266,6 +269,9 @@ message WriteSecuredCommand {
MxValue value = 5; 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 { message WriteSecured2Command {
int32 server_handle = 1; int32 server_handle = 1;
int32 item_handle = 2; int32 item_handle = 2;
@@ -525,6 +531,18 @@ message MxCommandReply {
// transport failures. // transport failures.
optional int32 hresult = 5; optional int32 hresult = 5;
MxValue return_value = 6; 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; repeated MxStatusProxy statuses = 7;
string diagnostic_message = 8; string diagnostic_message = 8;
@@ -225,6 +225,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
options.PipeConnectAttemptTimeoutMilliseconds, options.PipeConnectAttemptTimeoutMilliseconds,
"MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds must be greater than zero.", "MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds must be greater than zero.",
builder); builder);
if (options.WriteCompletionWaitMilliseconds < 0)
{
builder.Add("MxGateway:Worker:WriteCompletionWaitMilliseconds must be greater than or equal to zero.");
}
AddIfNotPositive( AddIfNotPositive(
options.ShutdownTimeoutSeconds, options.ShutdownTimeoutSeconds,
"MxGateway:Worker:ShutdownTimeoutSeconds must be greater than zero.", "MxGateway:Worker:ShutdownTimeoutSeconds must be greater than zero.",
@@ -24,6 +24,15 @@ public sealed class WorkerOptions
/// <summary>The timeout in milliseconds for connecting to the worker pipe.</summary> /// <summary>The timeout in milliseconds for connecting to the worker pipe.</summary>
public int PipeConnectAttemptTimeoutMilliseconds { get; init; } = 2000; public int PipeConnectAttemptTimeoutMilliseconds { get; init; } = 2000;
/// <summary>
/// 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.
/// </summary>
public int WriteCompletionWaitMilliseconds { get; init; } = 1500;
/// <summary>The maximum time in seconds for graceful shutdown.</summary> /// <summary>The maximum time in seconds for graceful shutdown.</summary>
public int ShutdownTimeoutSeconds { get; init; } = 10; public int ShutdownTimeoutSeconds { get; init; } = 10;
@@ -21,6 +21,14 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName = public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName =
"MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS"; "MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS";
/// <summary>
/// Conveys MxGateway:Worker:WriteCompletionWaitMilliseconds to the worker:
/// the bounded wait for the OnWriteComplete callback on
/// WriteSecured/WriteSecured2 replies. 0 disables the wait.
/// </summary>
public const string WorkerWriteCompletionWaitEnvironmentVariableName =
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
private readonly IWorkerProcessFactory _processFactory; private readonly IWorkerProcessFactory _processFactory;
private readonly IWorkerStartupProbe _startupProbe; private readonly IWorkerStartupProbe _startupProbe;
private readonly GatewayMetrics _metrics; private readonly GatewayMetrics _metrics;
@@ -175,6 +183,8 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
startInfo.Environment[WorkerNonceEnvironmentVariableName] = request.Nonce; startInfo.Environment[WorkerNonceEnvironmentVariableName] = request.Nonce;
startInfo.Environment[WorkerPipeConnectAttemptTimeoutEnvironmentVariableName] = startInfo.Environment[WorkerPipeConnectAttemptTimeoutEnvironmentVariableName] =
_workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); _workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
_workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
commandLine = new WorkerProcessCommandLine(executablePath, arguments); commandLine = new WorkerProcessCommandLine(executablePath, arguments);
@@ -41,6 +41,7 @@ public sealed class GatewayOptionsTests
Assert.Equal(3, options.Worker.StartupProbeRetryAttempts); Assert.Equal(3, options.Worker.StartupProbeRetryAttempts);
Assert.Equal(250, options.Worker.StartupProbeRetryDelayMilliseconds); Assert.Equal(250, options.Worker.StartupProbeRetryDelayMilliseconds);
Assert.Equal(2000, options.Worker.PipeConnectAttemptTimeoutMilliseconds); Assert.Equal(2000, options.Worker.PipeConnectAttemptTimeoutMilliseconds);
Assert.Equal(1500, options.Worker.WriteCompletionWaitMilliseconds);
Assert.Equal(10, options.Worker.ShutdownTimeoutSeconds); Assert.Equal(10, options.Worker.ShutdownTimeoutSeconds);
Assert.Equal(5, options.Worker.HeartbeatIntervalSeconds); Assert.Equal(5, options.Worker.HeartbeatIntervalSeconds);
Assert.Equal(15, options.Worker.HeartbeatGraceSeconds); 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: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: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: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: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: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).")] [InlineData("MxGateway:Sessions:DetachGraceSeconds", "-1", "MxGateway:Sessions:DetachGraceSeconds must be zero or greater (0 disables detach-grace retention).")]
@@ -43,6 +43,10 @@ public sealed class WorkerProcessLauncherTests
"2000", "2000",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerPipeConnectAttemptTimeoutEnvironmentVariableName]); WorkerProcessLauncher.WorkerPipeConnectAttemptTimeoutEnvironmentVariableName]);
Assert.Equal(
"1500",
processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]);
Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal); Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal);
Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal); Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal);
Assert.False(pipeReservation.DisposeCalled); Assert.False(pipeReservation.DisposeCalled);
@@ -110,6 +110,47 @@ public sealed class MxAccessBaseEventSinkTests
Assert.Same(cache, sink.ValueCache); Assert.Same(cache, sink.ValueCache);
} }
/// <summary>
/// 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).
/// </summary>
[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<MXSTATUS_PROXY>();
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));
}
/// <summary>
/// Verifies that the sink-bound write-completion cache is exposed for sharing
/// with the owning <see cref="MxAccessSession"/> so the sink's recordings and
/// the write executor's waits see the same instance.
/// </summary>
[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);
}
/// <summary> /// <summary>
/// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences. /// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences.
/// </summary> /// </summary>
@@ -897,6 +897,9 @@ public sealed class MxAccessCommandExecutorTests
FakeMxAccessComObjectFactory factory = new(fakeComObject); FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime(); using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink()); 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); await session.StartAsync(workerProcessId: 1234);
MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand( MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
@@ -919,6 +922,8 @@ public sealed class MxAccessCommandExecutorTests
FakeMxAccessComObjectFactory factory = new(fakeComObject); FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime(); using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink()); 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); await session.StartAsync(workerProcessId: 1234);
DateTime timestamp = new(2026, 5, 19, 13, 30, 0, DateTimeKind.Utc); DateTime timestamp = new(2026, 5, 19, 13, 30, 0, DateTimeKind.Utc);
@@ -934,6 +939,192 @@ public sealed class MxAccessCommandExecutorTests
Assert.Equal(44, fakeComObject.WriteVerifierUserId); Assert.Equal(44, fakeComObject.WriteVerifierUserId);
} }
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxCommandReply> 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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// Verifies that WriteSecured2 correlates the same way as WriteSecured
/// (fast-completion edge).
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxCommandReply> 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<MxStatusProxy> CreateCompletionRows(int detail)
{
return new Google.Protobuf.Collections.RepeatedField<MxStatusProxy>
{
new MxStatusProxy
{
Success = 1,
Category = MxStatusCategory.Ok,
Detail = detail,
},
};
}
/// <summary>Verifies that Write without a payload returns an invalid request error.</summary> /// <summary>Verifies that Write without a payload returns an invalid request error.</summary>
/// <returns>A task that represents the asynchronous operation.</returns> /// <returns>A task that represents the asynchronous operation.</returns>
[Fact] [Fact]
@@ -1775,6 +1966,28 @@ public sealed class MxAccessCommandExecutorTests
TimeSpan.FromMilliseconds(25)); TimeSpan.FromMilliseconds(25));
} }
/// <summary>
/// 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
/// <see cref="MxAccessSession.Create"/> shares this cache with the write
/// executor, letting tests record completions the executor's bounded
/// wait then observes.
/// </summary>
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 sealed class FakeMxAccessComObject : IMxAccessServer
{ {
private readonly int registerHandle; private readonly int registerHandle;
@@ -1790,6 +2003,14 @@ public sealed class MxAccessCommandExecutorTests
private readonly IReadOnlyDictionary<int, Exception> writeExceptionByItemHandle; private readonly IReadOnlyDictionary<int, Exception> writeExceptionByItemHandle;
private readonly List<string> operationNames = new(); private readonly List<string> operationNames = new();
/// <summary>
/// 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.
/// </summary>
public Action? OnWriteSecuredCallback { get; set; }
/// <summary>Initializes a fake MXAccess COM object with the given handles and optional exceptions.</summary> /// <summary>Initializes a fake MXAccess COM object with the given handles and optional exceptions.</summary>
/// <param name="registerHandle">Return value for Register method.</param> /// <param name="registerHandle">Return value for Register method.</param>
/// <param name="addItemHandle">Return value for AddItem method.</param> /// <param name="addItemHandle">Return value for AddItem method.</param>
@@ -2100,6 +2321,7 @@ public sealed class MxAccessCommandExecutorTests
WriteValue = value; WriteValue = value;
WriteThreadId = Environment.CurrentManagedThreadId; WriteThreadId = Environment.CurrentManagedThreadId;
ThrowIfWriteFailureConfigured(itemHandle); ThrowIfWriteFailureConfigured(itemHandle);
OnWriteSecuredCallback?.Invoke();
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -2120,6 +2342,7 @@ public sealed class MxAccessCommandExecutorTests
WriteTimestamp = timestamp; WriteTimestamp = timestamp;
WriteThreadId = Environment.CurrentManagedThreadId; WriteThreadId = Environment.CurrentManagedThreadId;
ThrowIfWriteFailureConfigured(itemHandle); ThrowIfWriteFailureConfigured(itemHandle);
OnWriteSecuredCallback?.Invoke();
} }
private void ThrowIfWriteFailureConfigured(int itemHandle) private void ThrowIfWriteFailureConfigured(int itemHandle)
@@ -15,6 +15,48 @@ namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
/// </summary> /// </summary>
public sealed class MxAccessStaSessionTests public sealed class MxAccessStaSessionTests
{ {
/// <summary>
/// Verifies the launcher-env-var parse branches of
/// <see cref="MxAccessStaSession.ResolveWriteCompletionTimeout"/>:
/// 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.
/// </summary>
/// <param name="rawValue">Raw env-var value, or null for unset.</param>
/// <param name="expectedMilliseconds">Expected resolved wait, or null for the executor default.</param>
[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);
}
}
/// <summary> /// <summary>
/// Verifies that StartAsync creates the MXAccess COM object and attaches the event sink on the STA thread. /// Verifies that StartAsync creates the MXAccess COM object and attaches the event sink on the STA thread.
/// </summary> /// </summary>
@@ -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;
/// <summary>
/// Unit tests for <see cref="MxAccessWriteCompletionCache"/>. 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.
/// </summary>
public sealed class MxAccessWriteCompletionCacheTests
{
/// <summary>Verifies that Record bumps the version per key and keys stay isolated.</summary>
[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));
}
/// <summary>Verifies that a completion newer than the baseline is returned with its status rows.</summary>
[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<MxStatusProxy> statuses);
Assert.True(found);
MxStatusProxy row = Assert.Single(statuses);
Assert.Equal(4321, row.Detail);
Assert.Equal(MxStatusCategory.Ok, row.Category);
}
/// <summary>
/// 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.
/// </summary>
[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<MxStatusProxy> statuses);
Assert.False(found);
Assert.Empty(statuses);
}
/// <summary>
/// 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.
/// </summary>
[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<MxStatusProxy> statuses);
Assert.True(found);
Assert.True(pumpCalls >= 2);
Assert.Equal(55, Assert.Single(statuses).Detail);
}
/// <summary>Verifies that Record stores an independent clone of the caller's rows.</summary>
[Fact]
public void Record_ClonesStatuses()
{
MxAccessWriteCompletionCache cache = new();
RepeatedField<MxStatusProxy> 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<MxStatusProxy> statuses));
Assert.Equal(77, Assert.Single(statuses).Detail);
}
private static RepeatedField<MxStatusProxy> BuildStatuses(int detail)
{
return new RepeatedField<MxStatusProxy>
{
new MxStatusProxy
{
Success = 1,
Category = MxStatusCategory.Ok,
Detail = detail,
},
};
}
}
@@ -0,0 +1,15 @@
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Exposes the per-session <see cref="MxAccessWriteCompletionCache"/> an
/// event sink populates from OnWriteComplete callbacks, so
/// <see cref="MxAccessSession.Create"/> can share one instance between the
/// sink (writer) and the write command executor (reader). Implemented by
/// <see cref="MxAccessBaseEventSink"/> and by test sinks that cannot
/// attach to a live MXAccess COM object.
/// </summary>
public interface IWriteCompletionCacheProvider
{
/// <summary>The completion cache bound to this sink.</summary>
MxAccessWriteCompletionCache WriteCompletionCache { get; }
}
@@ -5,11 +5,12 @@ using Proto = ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>Sink for MXAccess COM events that converts them to protobuf format.</summary> /// <summary>Sink for MXAccess COM events that converts them to protobuf format.</summary>
public sealed class MxAccessBaseEventSink : IMxAccessEventSink public sealed class MxAccessBaseEventSink : IMxAccessEventSink, IWriteCompletionCacheProvider
{ {
private readonly MxAccessEventMapper eventMapper; private readonly MxAccessEventMapper eventMapper;
private readonly MxAccessEventQueue eventQueue; private readonly MxAccessEventQueue eventQueue;
private readonly MxAccessValueCache valueCache; private readonly MxAccessValueCache valueCache;
private readonly MxAccessWriteCompletionCache writeCompletionCache;
private LMXProxyServerClass? server; private LMXProxyServerClass? server;
private string sessionId = string.Empty; private string sessionId = string.Empty;
@@ -50,10 +51,32 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
MxAccessEventQueue eventQueue, MxAccessEventQueue eventQueue,
MxAccessEventMapper eventMapper, MxAccessEventMapper eventMapper,
MxAccessValueCache valueCache) MxAccessValueCache valueCache)
: this(eventQueue, eventMapper, valueCache, new MxAccessWriteCompletionCache())
{
}
/// <summary>
/// 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
/// <c>OnWriteComplete</c> dispatch so the worker's write executor can
/// correlate a WriteSecured/WriteSecured2 reply with the MXAccess
/// completion outcome.
/// </summary>
/// <param name="eventQueue">Queue for buffering converted MXAccess events.</param>
/// <param name="eventMapper">Converter for MXAccess events to protobuf format.</param>
/// <param name="valueCache">Per-session last-value cache shared with the MxAccessSession.</param>
/// <param name="writeCompletionCache">Per-session OnWriteComplete cache shared with the MxAccessSession.</param>
public MxAccessBaseEventSink(
MxAccessEventQueue eventQueue,
MxAccessEventMapper eventMapper,
MxAccessValueCache valueCache,
MxAccessWriteCompletionCache writeCompletionCache)
{ {
this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue)); this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper)); this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache)); this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
} }
/// <summary> /// <summary>
@@ -62,6 +85,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
/// </summary> /// </summary>
public MxAccessValueCache ValueCache => valueCache; public MxAccessValueCache ValueCache => valueCache;
/// <summary>
/// The OnWriteComplete completion cache populated by this sink. Exposed
/// via <see cref="IWriteCompletionCacheProvider"/> so the
/// MxAccessSession can share the same instance with the write command
/// executor's bounded completion wait.
/// </summary>
public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;
/// <inheritdoc /> /// <inheritdoc />
public void Attach( public void Attach(
object mxAccessComObject, object mxAccessComObject,
@@ -143,11 +174,18 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
ref MXSTATUS_PROXY[] pVars) ref MXSTATUS_PROXY[] pVars)
{ {
MXSTATUS_PROXY[] statuses = pVars; MXSTATUS_PROXY[] statuses = pVars;
EnqueueEvent(() => eventMapper.CreateOnWriteComplete( // 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, sessionId,
hLMXServerHandle, hLMXServerHandle,
phItemHandle, phItemHandle,
statuses)); statuses),
mxEvent => writeCompletionCache.Record(hLMXServerHandle, phItemHandle, mxEvent.Statuses));
} }
/// <summary> /// <summary>
@@ -14,11 +14,22 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
/// <summary>Default per-tag timeout used when <c>ReadBulkCommand.timeout_ms</c> is zero.</summary> /// <summary>Default per-tag timeout used when <c>ReadBulkCommand.timeout_ms</c> is zero.</summary>
internal static readonly TimeSpan DefaultReadBulkTimeout = TimeSpan.FromMilliseconds(1000); 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 MxAccessSession session;
private readonly VariantConverter variantConverter; private readonly VariantConverter variantConverter;
private readonly MxStatusProxyConverter statusProxyConverter; private readonly MxStatusProxyConverter statusProxyConverter;
private readonly IAlarmCommandHandler? alarmCommandHandler; private readonly IAlarmCommandHandler? alarmCommandHandler;
private readonly Action pumpStep; private readonly Action pumpStep;
private readonly TimeSpan writeCompletionTimeout;
/// <summary> /// <summary>
/// Initializes a command executor with an MXAccess session. /// 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="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
/// <param name="alarmCommandHandler">Optional handler for alarm-side commands.</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="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( public MxAccessCommandExecutor(
MxAccessSession session, MxAccessSession session,
VariantConverter variantConverter, VariantConverter variantConverter,
IAlarmCommandHandler? alarmCommandHandler, IAlarmCommandHandler? alarmCommandHandler,
Action? pumpStep) Action? pumpStep,
TimeSpan? writeCompletionTimeout = null)
{ {
this.session = session ?? throw new ArgumentNullException(nameof(session)); this.session = session ?? throw new ArgumentNullException(nameof(session));
this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter)); this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter));
this.statusProxyConverter = new MxStatusProxyConverter(); this.statusProxyConverter = new MxStatusProxyConverter();
this.alarmCommandHandler = alarmCommandHandler; this.alarmCommandHandler = alarmCommandHandler;
this.pumpStep = pumpStep ?? (static () => { }); this.pumpStep = pumpStep ?? (static () => { });
this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout;
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -457,6 +477,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
return CreateInvalidRequestReply(command, "WriteSecured command value is required."); 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( session.WriteSecured(
writeSecuredCommand.ServerHandle, writeSecuredCommand.ServerHandle,
writeSecuredCommand.ItemHandle, writeSecuredCommand.ItemHandle,
@@ -464,7 +492,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
writeSecuredCommand.VerifierUserId, writeSecuredCommand.VerifierUserId,
variantConverter.ConvertToComValue(writeSecuredCommand.Value)); 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) private MxCommandReply ExecuteWriteSecured2(StaCommand command)
@@ -485,6 +520,12 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
return CreateInvalidRequestReply(command, "WriteSecured2 command timestamp value is required."); 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( session.WriteSecured2(
writeSecured2Command.ServerHandle, writeSecured2Command.ServerHandle,
writeSecured2Command.ItemHandle, writeSecured2Command.ItemHandle,
@@ -493,7 +534,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
variantConverter.ConvertToComValue(writeSecured2Command.Value), variantConverter.ConvertToComValue(writeSecured2Command.Value),
variantConverter.ConvertToComValue(writeSecured2Command.TimestampValue)); 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) 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) private static MxCommandReply CreateAlarmFailureReply(StaCommand command, Exception exception)
{ {
return new MxCommandReply return new MxCommandReply
@@ -13,6 +13,7 @@ public sealed class MxAccessSession : IDisposable
private readonly IMxAccessEventSink eventSink; private readonly IMxAccessEventSink eventSink;
private readonly MxAccessHandleRegistry handleRegistry; private readonly MxAccessHandleRegistry handleRegistry;
private readonly MxAccessValueCache valueCache; private readonly MxAccessValueCache valueCache;
private readonly MxAccessWriteCompletionCache writeCompletionCache;
private bool disposed; private bool disposed;
private MxAccessSession( private MxAccessSession(
@@ -21,6 +22,7 @@ public sealed class MxAccessSession : IDisposable
IMxAccessEventSink eventSink, IMxAccessEventSink eventSink,
MxAccessHandleRegistry handleRegistry, MxAccessHandleRegistry handleRegistry,
MxAccessValueCache valueCache, MxAccessValueCache valueCache,
MxAccessWriteCompletionCache writeCompletionCache,
int creationThreadId) int creationThreadId)
{ {
this.mxAccessComObject = mxAccessComObject ?? throw new ArgumentNullException(nameof(mxAccessComObject)); 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.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
this.handleRegistry = handleRegistry ?? throw new ArgumentNullException(nameof(handleRegistry)); this.handleRegistry = handleRegistry ?? throw new ArgumentNullException(nameof(handleRegistry));
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache)); this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
CreationThreadId = creationThreadId; CreationThreadId = creationThreadId;
} }
@@ -45,6 +48,14 @@ public sealed class MxAccessSession : IDisposable
/// </summary> /// </summary>
public MxAccessValueCache ValueCache => valueCache; public MxAccessValueCache ValueCache => valueCache;
/// <summary>
/// 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.
/// </summary>
public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;
/// <summary>Creates a WorkerReady message with session metadata.</summary> /// <summary>Creates a WorkerReady message with session metadata.</summary>
/// <param name="workerProcessId">Process ID of the worker.</param> /// <param name="workerProcessId">Process ID of the worker.</param>
/// <returns>The populated <see cref="WorkerReady"/> message.</returns> /// <returns>The populated <see cref="WorkerReady"/> message.</returns>
@@ -105,6 +116,9 @@ public sealed class MxAccessSession : IDisposable
eventSink, eventSink,
handleRegistry ?? new MxAccessHandleRegistry(), handleRegistry ?? new MxAccessHandleRegistry(),
valueCache ?? new MxAccessValueCache(), valueCache ?? new MxAccessValueCache(),
eventSink is IWriteCompletionCacheProvider provider
? provider.WriteCompletionCache
: new MxAccessWriteCompletionCache(),
creationThreadId ?? Environment.CurrentManagedThreadId); creationThreadId ?? Environment.CurrentManagedThreadId);
} }
@@ -149,12 +163,22 @@ public sealed class MxAccessSession : IDisposable
? baseSink.ValueCache ? baseSink.ValueCache
: new MxAccessValueCache(); : 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( return new MxAccessSession(
mxAccessComObject, mxAccessComObject,
new MxAccessComServer(mxAccessComObject), new MxAccessComServer(mxAccessComObject),
eventSink, eventSink,
new MxAccessHandleRegistry(), new MxAccessHandleRegistry(),
valueCache, valueCache,
writeCompletionCache,
Environment.CurrentManagedThreadId); Environment.CurrentManagedThreadId);
} }
catch (Exception exception) catch (Exception exception)
@@ -11,6 +11,14 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
public sealed class MxAccessStaSession : IWorkerRuntimeSession 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 static readonly TimeSpan AlarmPollInterval = TimeSpan.FromMilliseconds(500);
private readonly IMxAccessComObjectFactory factory; private readonly IMxAccessComObjectFactory factory;
@@ -157,6 +165,32 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// </summary> /// </summary>
public MxAccessEventQueue EventQueue => eventQueue; 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> /// <summary>
/// Starts the MXAccess COM session asynchronously. /// Starts the MXAccess COM session asynchronously.
/// </summary> /// </summary>
@@ -208,12 +242,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
session, session,
new VariantConverter(), new VariantConverter(),
alarmCommandHandler, 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); return session.CreateWorkerReady(workerProcessId);
}, },
@@ -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;
/// <summary>
/// Per-session cache of the most recent <c>OnWriteComplete</c> 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.
/// </summary>
/// <remarks>
/// Same threading posture as <see cref="MxAccessValueCache"/>: 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.
/// </remarks>
public sealed class MxAccessWriteCompletionCache
{
private readonly Dictionary<long, CompletionEntry> entries = new();
private readonly object syncRoot = new();
/// <summary>Records the status rows of a fresh OnWriteComplete callback for the given handle pair.</summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle.</param>
/// <param name="statuses">Status rows from the mapped OnWriteComplete event; cloned before storing.</param>
public void Record(
int serverHandle,
int itemHandle,
RepeatedField<MxStatusProxy> 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());
}
}
/// <summary>Returns the current completion version for a handle pair, or 0 if none was recorded.</summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle.</param>
/// <returns>The current completion version, or 0 if no completion was recorded.</returns>
public ulong CurrentVersion(
int serverHandle,
int itemHandle)
{
lock (syncRoot)
{
return entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry existing)
? existing.Version
: 0UL;
}
}
/// <summary>
/// Polls for a completion newer than <paramref name="sinceVersion"/> until it
/// arrives or the deadline elapses, calling <paramref name="pumpStep"/> on every
/// poll iteration so the worker's STA can dispatch the inbound MXAccess
/// OnWriteComplete message. Same loop shape as
/// <see cref="MxAccessValueCache.TryWaitForUpdate"/>.
/// </summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle.</param>
/// <param name="sinceVersion">Version snapshot captured before the write COM call.</param>
/// <param name="deadlineUtc">Absolute UTC deadline.</param>
/// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
/// <param name="statuses">The recorded status rows if a completion arrived before the deadline; empty otherwise.</param>
/// <param name="pollIntervalMs">How long to sleep between pump cycles. Default 5 ms.</param>
/// <returns><see langword="true"/> if a completion newer than <paramref name="sinceVersion"/> arrived before the deadline; otherwise <see langword="false"/>.</returns>
public bool TryWaitForCompletion(
int serverHandle,
int itemHandle,
ulong sinceVersion,
DateTime deadlineUtc,
Action pumpStep,
out RepeatedField<MxStatusProxy> 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<MxStatusProxy>();
return false;
}
Thread.Sleep(pollIntervalMs);
}
}
private static long CreateItemKey(
int serverHandle,
int itemHandle)
{
return ((long)serverHandle << 32) | (uint)itemHandle;
}
/// <summary>
/// Snapshot of the most recent OnWriteComplete status rows for a handle
/// pair. <see cref="Version"/> increments by one on every
/// <see cref="Record"/> call so the write executor can detect "a new
/// completion arrived since I captured my baseline".
/// </summary>
/// <remarks>
/// Plain readonly struct (not a record) so this compiles under the
/// worker's net48 target, which lacks <c>IsExternalInit</c>.
/// </remarks>
private readonly struct CompletionEntry
{
public CompletionEntry(
ulong version,
RepeatedField<MxStatusProxy> statuses)
{
Version = version;
Statuses = statuses;
}
public ulong Version { get; }
public RepeatedField<MxStatusProxy> Statuses { get; }
}
}