Compare commits
10 Commits
129e47e541
...
2b468bd8fc
| Author | SHA1 | Date | |
|---|---|---|---|
| 2b468bd8fc | |||
| 431a096cab | |||
| b0e2b8ba74 | |||
| 66fe063410 | |||
| 8de23086d0 | |||
| a76ecdd59c | |||
| fc23a65cca | |||
| aec95b78c9 | |||
| f9229ee44d | |||
| 5dbe93d13e |
@@ -534,6 +534,48 @@ against the live MXAccess attribute set.
|
||||
- [Alarm Client Discovery — Subtag provider](./AlarmClientDiscovery.md)
|
||||
- [gRPC Contract — provider_status and degraded fields](./Grpc.md)
|
||||
|
||||
## Secured-Write Completion Correlation
|
||||
|
||||
MXAccess writes are fire-and-forget: the toolkit call returns before the
|
||||
Galaxy commit, and the per-item outcome only exists in the later
|
||||
`OnWriteComplete` COM callback. The original unary write reply therefore
|
||||
proved worker-side command acceptance only, forcing consumers (OtOpcUa's
|
||||
GalaxyDriver) to report every write as provisionally good.
|
||||
|
||||
For `WriteSecured`/`WriteSecured2` the worker now holds the unary reply for a
|
||||
bounded window (`MxGateway:Worker:WriteCompletionWaitMilliseconds`, default
|
||||
1.5 s, `0` disables; conveyed to the worker via
|
||||
`MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS`) and copies the matching
|
||||
callback's status rows onto `MxCommandReply.statuses`. Key choices, argued in
|
||||
[the design doc](./plans/2026-08-09-write-completion-correlation-design.md):
|
||||
|
||||
- **Pump-wait on the STA, not a parked reply.** The executor holds the STA
|
||||
thread but pumps Windows messages each poll — the shipped ReadBulk pattern —
|
||||
because commands serialize per session anyway, so freeing the STA during the
|
||||
wait buys nothing and a parked reply would change the dispatcher/pipe
|
||||
contracts.
|
||||
- **Version baseline before the COM call** closes the fast-completion edge: a
|
||||
callback that dispatches while `WriteSecured` is still on the stack still
|
||||
correlates.
|
||||
- **Timeout returns today's shape** (protocol OK, empty statuses):
|
||||
unconfirmed is honest; a synthesized failure row would trigger consumer-side
|
||||
write-revert logic on slow-but-successful commits. The 1.5 s default stays
|
||||
inside OtOpcUa's 2 s Tier A write-resilience budget.
|
||||
- **Parity preserved.** `protocol_status`/`hresult` keep describing
|
||||
acceptance; the MXAccess outcome (success or failure) rides only in
|
||||
`statuses[0]`; the `OnWriteComplete` event still streams unchanged (nothing
|
||||
swallowed, nothing synthesized).
|
||||
- **Scope: secured writes only.** Plain `Write`/`Write2` and bulk writes stay
|
||||
fire-and-forget — the wait would add a device round-trip per write to
|
||||
high-rate supervisory loops.
|
||||
- **Best-effort correlation.** The callback carries only
|
||||
`(hItem, statuses)` — no transaction id — so concurrent writes to the same
|
||||
item within the window can swap rows; benign for the serialized single-write
|
||||
consumer contract.
|
||||
- **Client cancellation needs no special path**: a caller abandoning the RPC
|
||||
mid-wait leaves the worker to finish its bounded wait and reply; the gateway
|
||||
discards the reply, the session is never faulted.
|
||||
|
||||
## Later Revisit Items
|
||||
|
||||
These are explicit post-v1 revisit items, not open blockers:
|
||||
|
||||
@@ -114,6 +114,7 @@ launch CWD (SEC-01, SEC-33).
|
||||
| `MxGateway:Worker:StartupProbeRetryAttempts` | `3` | Number of retry attempts for transient worker startup probe failures before pipe connection and handshake continue. |
|
||||
| `MxGateway:Worker:StartupProbeRetryDelayMilliseconds` | `250` | Delay between transient startup probe retry attempts. |
|
||||
| `MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds` | `2000` | Per-attempt timeout used by the worker named-pipe connect retry path. The overall pipe connection still stays under the startup budget. |
|
||||
| `MxGateway:Worker:WriteCompletionWaitMilliseconds` | `1500` | Bounded wait the worker holds a `WriteSecured`/`WriteSecured2` reply for the matching MXAccess `OnWriteComplete` callback, so the reply's `statuses` carry the real commit outcome. `0` disables the wait (pure fire-and-forget replies). Must be `>= 0`. The gateway conveys the value to the worker via the `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` environment variable. Consumers that time their own writes must budget above this wait: OtOpcUa's GalaxyDriver wraps gateway writes in a 2 s Tier A resilience timeout, so a deployment raising this option past ~2000 must raise that driver `ResilienceConfig` write timeout in step or slow-but-successful commits surface as consumer-side failures. |
|
||||
| `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. |
|
||||
| `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. |
|
||||
| `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. |
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
# WriteSecured Completion Correlation — Design
|
||||
|
||||
Date: 2026-08-09
|
||||
Requested by: OtOpcUa (archreview finding 06/S-1, cross-repo residual — gateway half)
|
||||
Status: approved for implementation (peer contract confirmed over cross-session message)
|
||||
|
||||
## Problem
|
||||
|
||||
The unary `Invoke` reply for a `WriteSecured` / `WriteSecured2` command proves worker-side
|
||||
command *acceptance* only. MXAccess writes are fire-and-forget at the toolkit level: the
|
||||
real per-item outcome arrives later in the `OnWriteComplete` COM callback. Today the worker
|
||||
returns `CreateOkReply` immediately after the COM call, so `MxCommandReply.statuses` is
|
||||
always empty and consumers (OtOpcUa's `GatewayGalaxyDataWriter.TranslateReply`) must treat
|
||||
every write as provisionally good (`galaxy.writes.unconfirmed` meter).
|
||||
|
||||
Consumer contract (already shipped OtOpcUa-side): `statuses.Count > 0` → map `statuses[0]`
|
||||
through the MX status map to a real OPC UA StatusCode; empty → provisional Good +
|
||||
unconfirmed meter. The `statuses` field already exists on `MxCommandReply` (field 7); no
|
||||
proto shape change is needed.
|
||||
|
||||
## Approaches Considered
|
||||
|
||||
1. **Executor pump-wait + versioned completion cache (chosen).** After the `WriteSecured`
|
||||
COM call, the executor holds the STA thread but explicitly pumps Windows messages each
|
||||
poll iteration until the matching completion is recorded or a bounded deadline passes.
|
||||
This is exactly the shipped `ReadBulk` pattern (`MxAccessValueCache.TryWaitForUpdate` +
|
||||
`pumpStep` → `StaRuntime.PumpPendingMessages()`), so it adds no new threading model.
|
||||
2. **Parked asynchronous reply.** Executor returns a "pending" marker; the dispatcher parks
|
||||
the correlation and the pipe reply is written later from the event dispatch. Rejected:
|
||||
changes the `IStaCommandExecutor`/dispatcher/pipe contracts for no real gain — commands
|
||||
serialize per session anyway, so freeing the STA during the wait buys nothing.
|
||||
3. **Gateway-side correlation.** Gateway watches the session event stream for the
|
||||
`OnWriteComplete` after the worker reply. Rejected: races the event drain cadence,
|
||||
couples the gateway to event semantics, still holds the unary RPC, and spreads the
|
||||
feature across two processes.
|
||||
|
||||
## Design
|
||||
|
||||
All changes are worker-side (`ZB.MOM.WW.MxGateway.Worker`, net48 x86). The gateway's
|
||||
`Invoke` already forwards the worker `MxCommandReply` (statuses included) verbatim.
|
||||
|
||||
### New: `MxAccessWriteCompletionCache`
|
||||
|
||||
Mirror of `MxAccessValueCache`, keyed by `(serverHandle, itemHandle)` (packed long), one
|
||||
entry per key holding the most recent completion's `RepeatedField<MxStatusProxy>` (cloned)
|
||||
plus a monotonically increasing per-key `Version`. API:
|
||||
|
||||
- `Record(int serverHandle, int itemHandle, RepeatedField<MxStatusProxy> statuses)`
|
||||
- `ulong CurrentVersion(int serverHandle, int itemHandle)` — 0 when absent
|
||||
- `bool TryWaitForCompletion(int serverHandle, int itemHandle, ulong sinceVersion,
|
||||
DateTime deadlineUtc, Action pumpStep, out RepeatedField<MxStatusProxy> statuses,
|
||||
int pollIntervalMs = 5)` — pump/poll loop identical in shape to
|
||||
`MxAccessValueCache.TryWaitForUpdate`.
|
||||
|
||||
Same locking posture as the value cache: everything runs on the STA thread; a sync root
|
||||
keeps it nominally thread-safe for tests.
|
||||
|
||||
### Sink: record completions
|
||||
|
||||
`MxAccessBaseEventSink` owns a `MxAccessWriteCompletionCache` (new optional ctor param,
|
||||
exposed as a property) and its `OnWriteComplete` handler records into it via the existing
|
||||
`EnqueueEvent` post-publish hook (same pattern as the value cache on `OnDataChange`):
|
||||
the streamed `MxEvent` is built exactly once by the mapper, enqueued unchanged for the
|
||||
event stream, and its `Statuses` are then recorded into the cache. The event stream is
|
||||
not altered — nothing is swallowed or synthesized.
|
||||
|
||||
A new seam interface `IWriteCompletionCacheProvider { MxAccessWriteCompletionCache
|
||||
WriteCompletionCache { get; } }` is implemented by `MxAccessBaseEventSink` and by test
|
||||
sinks. `MxAccessSession.Create` pulls the cache from the sink through that interface
|
||||
(fallback: fresh instance), mirroring the existing `ValueCache` sharing, and exposes it
|
||||
on the session.
|
||||
|
||||
### Executor: bounded pump-wait
|
||||
|
||||
In `ExecuteWriteSecured` / `ExecuteWriteSecured2`:
|
||||
|
||||
1. Capture `baseline = cache.CurrentVersion(serverHandle, itemHandle)` **before** the COM
|
||||
call — this closes the fast-completion ordering edge: a callback that dispatches during
|
||||
or immediately after the COM call bumps the version past the baseline and still
|
||||
correlates.
|
||||
2. Call `session.WriteSecured(...)` as today.
|
||||
3. `cache.TryWaitForCompletion(..., baseline, deadline, pumpStep, out statuses)`; on
|
||||
success, `reply.Statuses.Add(statuses)`; on timeout, return the reply exactly as today
|
||||
(protocol OK, empty statuses) — the consumer's honest-unconfirmed path. No invented
|
||||
failure rows.
|
||||
|
||||
The reply's `ProtocolStatus`/`Hresult` stay untouched by the completion outcome: the
|
||||
command was accepted; the MX outcome (success *or* failure) is carried only in
|
||||
`statuses[0]`. That preserves MXAccess parity (the native API returns void; the outcome
|
||||
exists only in the callback) while enriching the reply with information that was already
|
||||
on the wire.
|
||||
|
||||
Timeout default: **1.5 s** (`MxAccessCommandExecutor.DefaultWriteCompletionTimeout`).
|
||||
The consumer-side budget drives this: OtOpcUa wraps the driver write in a Tier A
|
||||
resilience policy with a 2 s timeout and a 5-failure breaker, so a worker wait longer
|
||||
than 2 s would convert slow-but-successful commits into consumer-side false failures
|
||||
(node revert + breaker pressure). 1.5 s covers the common fast-commit case and degrades
|
||||
a slow commit to the honest-unconfirmed path instead. It also sits well under the
|
||||
gateway's 30 s `DefaultCommandTimeoutSeconds` IPC wait and the STA watchdog's 75 s
|
||||
dispatched-command ceiling.
|
||||
|
||||
The wait is deployment-configurable end to end, following the existing
|
||||
pipe-connect-timeout pattern: a new gateway option
|
||||
`MxGateway:Worker:WriteCompletionWaitMilliseconds` (default `1500`, validated `>= 0`;
|
||||
`0` disables the wait and restores pure fire-and-forget replies) is exported by
|
||||
`WorkerProcessLauncher` as the `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` environment
|
||||
variable, which the worker reads at session construction. A deployment that raises the
|
||||
gateway wait must raise the OtOpcUa driver's Write `ResilienceConfig` timeout in step
|
||||
(per-instance operator config on the OtOpcUa side) — documented in
|
||||
`docs/GatewayConfiguration.md`. `MxAccessStaSession` additionally gets an internal
|
||||
`WriteCompletionTimeout` seam (read when it constructs the executor in `StartAsync`) so
|
||||
tests can shorten it without env plumbing.
|
||||
|
||||
Client cancellation mid-wait needs no new code path, only verification: when the caller
|
||||
cancels the unary RPC, the gateway abandons its IPC reply wait, but the worker command
|
||||
is already in flight — `CancelCommand` only dequeues *queued* commands. The executor
|
||||
simply finishes its bounded wait and replies; the gateway discards the reply. The
|
||||
session is never faulted and the `OnWriteComplete` event still flows on the stream.
|
||||
|
||||
### Scope
|
||||
|
||||
- **In**: `WriteSecured`, `WriteSecured2` (single-item secured writes — inherently
|
||||
low-rate operator actions, and exactly the OtOpcUa single-write contract). Default-on.
|
||||
- **Out**: plain `Write`/`Write2` and all bulk write commands stay fire-and-forget —
|
||||
waiting would add a device round-trip of latency to high-rate supervisory write loops.
|
||||
- **Correlation fidelity is best-effort**: the MXAccess callback carries only
|
||||
`(hItem, statuses)` — no transaction id — so a concurrent write to the same item within
|
||||
the wait window can be attributed to the wrong writer (worst case two writes to the
|
||||
same item swap status rows — benign for the serialized single-write consumer
|
||||
contract). Documented on the proto field.
|
||||
|
||||
## Error handling
|
||||
|
||||
- Completion never arrives (device down): bounded 1.5 s wait, then today's reply shape.
|
||||
- Event queue overflow during completion: the queue records a fault and the fail-fast
|
||||
design tears the session down; the post-publish hook not firing in that case is moot.
|
||||
- COM call throws: unchanged — the dispatcher's existing exception path replies with the
|
||||
native HResult; no wait is entered.
|
||||
|
||||
## Testing (Worker.Tests, x86 — verified on windev)
|
||||
|
||||
- `MxAccessWriteCompletionCacheTests`: record/version monotonicity, wait success,
|
||||
deadline expiry, baseline-before-record fast-completion ordering.
|
||||
- `MxAccessBaseEventSinkTests`: `OnWriteComplete` both enqueues the event *and* records
|
||||
the completion; cache instance is the sink-bound one.
|
||||
- `MxAccessCommandExecutorTests` (via `MxAccessStaSession.DispatchAsync` + fake COM
|
||||
object + test sink implementing `IWriteCompletionCacheProvider`):
|
||||
- completion recorded synchronously inside the fake's `WriteSecured` (fast edge) →
|
||||
reply carries `statuses[0]`;
|
||||
- completion recorded from the test thread while the executor pump-waits → reply
|
||||
carries `statuses[0]`;
|
||||
- no completion + shortened timeout → protocol OK, empty statuses;
|
||||
- `WriteSecured2` mirrors; plain `Write` does not wait.
|
||||
- Gateway tests (macOS-runnable): `GatewayOptionsValidator` accepts `>= 0` and rejects
|
||||
negative `WriteCompletionWaitMilliseconds`; `WorkerProcessLauncher` exports
|
||||
`MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` from the option (mirroring the existing
|
||||
pipe-connect-timeout env assertion).
|
||||
|
||||
## Docs in the same change
|
||||
|
||||
- `mxaccess_gateway.proto`: comment on `MxCommandReply.statuses` and the
|
||||
`WriteSecuredCommand`/`WriteSecured2Command` messages describing the correlated
|
||||
completion contract (populated within the wait window; empty = unconfirmed;
|
||||
best-effort correlation). Comment-only → wire-identical; regenerate + commit
|
||||
`Contracts/Generated` (required for the C# build); other clients' generated code is
|
||||
functionally unchanged.
|
||||
- `docs/GatewayConfiguration.md`: the new `MxGateway:Worker:WriteCompletionWaitMilliseconds`
|
||||
option, including the pairing rule with the OtOpcUa driver's Write resilience timeout.
|
||||
- `gateway.md` command/event surface note; `docs/DesignDecisions.md` entry.
|
||||
@@ -0,0 +1,575 @@
|
||||
# WriteSecured Completion Correlation Implementation Plan
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** Populate `MxCommandReply.statuses[0]` on unary `WriteSecured`/`WriteSecured2` replies with the correlated MXAccess `OnWriteComplete` outcome, bounded by a configurable wait (default 1.5 s), falling back to today's empty-statuses shape on timeout.
|
||||
|
||||
**Architecture:** Worker-side only (plus one gateway config option). A versioned per-`(serverHandle, itemHandle)` completion cache (mirror of `MxAccessValueCache`) is populated by the event sink's `OnWriteComplete` post-publish hook; the STA command executor captures a version baseline before the COM call, then pump-waits (ReadBulk precedent) until a newer completion lands or the deadline passes. Design: `docs/plans/2026-08-09-write-completion-correlation-design.md`.
|
||||
|
||||
**Tech Stack:** .NET Framework 4.8 x86 worker (no init-only props/positional records!), .NET 10 gateway, protobuf via Grpc.Tools regen. Worker builds/tests run ONLY on windev (10.100.0.48) — local macOS verification covers the gateway + contracts.
|
||||
|
||||
---
|
||||
|
||||
### Task 0: Create feature branch
|
||||
|
||||
**Classification:** trivial
|
||||
**Estimated implement time:** ~1 min
|
||||
**Parallelizable with:** none
|
||||
|
||||
```bash
|
||||
cd /Users/dohertj2/Desktop/MxAccessGateway && git checkout -b feat/write-completion-correlation
|
||||
```
|
||||
|
||||
### Task 1: Proto contract comments + regen
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** none (everything builds on the regenerated contracts)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto` (~line 528 `statuses`, ~line 259 `WriteSecuredCommand`, ~line 269 `WriteSecured2Command`)
|
||||
- Regenerate: `src/ZB.MOM.WW.MxGateway.Contracts/Generated/*.cs`
|
||||
|
||||
**Step 1:** On `repeated MxStatusProxy statuses = 7;` in `MxCommandReply`, add above the field:
|
||||
|
||||
```proto
|
||||
// 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.
|
||||
```
|
||||
|
||||
**Step 2:** On `message WriteSecuredCommand` and `message WriteSecured2Command`, append to the existing leading comment (or add one): `// The unary reply's statuses field carries the correlated OnWriteComplete outcome when it arrives within the worker's bounded wait — see MxCommandReply.statuses.`
|
||||
|
||||
**Step 3:** Regenerate + verify wire-identical build:
|
||||
|
||||
```bash
|
||||
rm src/ZB.MOM.WW.MxGateway.Contracts/Generated/*.cs
|
||||
dotnet build src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj
|
||||
```
|
||||
Expected: build succeeds, `git diff --stat` shows only comment-churn in Generated.
|
||||
|
||||
**Step 4:** Commit: `git add src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto src/ZB.MOM.WW.MxGateway.Contracts/Generated && git commit -m "docs(proto): document the correlated write-completion statuses contract"`
|
||||
(Comment-only proto change is wire-identical; other clients' generated code is intentionally not regenerated — no functional delta.)
|
||||
|
||||
### Task 2: MxAccessWriteCompletionCache + tests
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 7
|
||||
|
||||
**Files:**
|
||||
- Create: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs`
|
||||
- Create: `src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs`
|
||||
|
||||
**Step 1:** Create the cache — mirror `MxAccessValueCache`'s shape, locking, and net48 constraints (no init-only, plain struct/class):
|
||||
|
||||
```csharp
|
||||
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.</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; }
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Step 2:** Tests (mirror `MxAccessValueCacheTests` style; build `MxStatusProxy` rows inline):
|
||||
|
||||
- `Record_IncrementsVersionPerKey` — two `Record` calls on the same pair → `CurrentVersion` 1 then 2; a different pair stays independent.
|
||||
- `TryWaitForCompletion_WhenCompletionNewerThanBaseline_ReturnsStatuses` — record once, wait with `sinceVersion: 0`, deadline in the future → `true`, statuses round-trip (assert an `MxStatusProxy` field value survives the clone).
|
||||
- `TryWaitForCompletion_WhenOnlyStaleCompletion_TimesOut` — record once, wait with `sinceVersion: CurrentVersion(...)` and a deadline ~50 ms out → `false`, out statuses empty.
|
||||
- `TryWaitForCompletion_InvokesPumpStepEachIteration` — pumpStep increments a counter; on the counter's second call, `Record` the completion (this proves the pump loop is what lets the callback land); assert `true` and counter >= 2.
|
||||
- `Record_ClonesStatuses` — mutate the caller's `RepeatedField` after `Record`; waited-out statuses unaffected.
|
||||
|
||||
**Step 3:** Cannot compile locally (worker is Windows-only) — defer build/test to Task 9 (windev). Commit: `git add src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs && git commit -m "feat(worker): versioned OnWriteComplete completion cache"`
|
||||
|
||||
### Task 3: Sink records completions (+ provider seam)
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** Task 7
|
||||
|
||||
**Files:**
|
||||
- Create: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs`
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs`
|
||||
- Test: `src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs`
|
||||
|
||||
**Step 1:** New seam interface:
|
||||
|
||||
```csharp
|
||||
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; }
|
||||
}
|
||||
```
|
||||
|
||||
**Step 2:** `MxAccessBaseEventSink` — declare `IWriteCompletionCacheProvider` on the class; add a `private readonly MxAccessWriteCompletionCache writeCompletionCache;` initialized in the widest ctor (add a new optional-most ctor overload following the existing chain pattern: the 3-arg `(eventQueue, eventMapper, valueCache)` ctor chains to a new 4-arg `(eventQueue, eventMapper, valueCache, writeCompletionCache)` with a fresh cache); expose `public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;`. Change `OnWriteComplete` to use the post-publish hook (same pattern as `OnDataChange`'s value-cache publish — post-publish only runs after the event cleared the queue, and a queue overflow faults the session anyway):
|
||||
|
||||
```csharp
|
||||
MXSTATUS_PROXY[] statuses = pVars;
|
||||
EnqueueEvent(
|
||||
() => eventMapper.CreateOnWriteComplete(
|
||||
sessionId,
|
||||
hLMXServerHandle,
|
||||
phItemHandle,
|
||||
statuses),
|
||||
mxEvent => writeCompletionCache.Record(hLMXServerHandle, phItemHandle, mxEvent.Statuses));
|
||||
```
|
||||
|
||||
**Step 3:** Tests in `MxAccessBaseEventSinkTests` (mirror `OnDataChange_ComCallback_PopulatesValueCache` and `ValueCache_ReturnsTheInstanceBoundAtConstruction`):
|
||||
- `OnWriteComplete_ComCallback_RecordsCompletionAndStillEnqueuesEvent` — drive `sink.OnWriteComplete(7, 21, ref proxies)`; assert the queue got the OnWriteComplete event AND `cache.CurrentVersion(7, 21) == 1`.
|
||||
- `WriteCompletionCache_ReturnsTheInstanceBoundAtConstruction`.
|
||||
|
||||
**Step 4:** Commit: `git commit -m "feat(worker): event sink records OnWriteComplete rows into the completion cache"` (explicit paths).
|
||||
|
||||
### Task 4: MxAccessSession plumbing
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~3 min
|
||||
**Parallelizable with:** Task 7
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessSession.cs` (private ctor ~line 18, `Create` ~line 145)
|
||||
|
||||
**Step 1:** Add `private readonly MxAccessWriteCompletionCache writeCompletionCache;` + ctor param (after `valueCache`) + null guard; add property mirroring `ValueCache`:
|
||||
|
||||
```csharp
|
||||
/// <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;
|
||||
```
|
||||
|
||||
**Step 2:** In `Create`, next to the value-cache sharing block:
|
||||
|
||||
```csharp
|
||||
// Share the sink's completion cache the same way (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();
|
||||
```
|
||||
|
||||
Pass it to the ctor.
|
||||
|
||||
**Step 3:** Commit: `git commit -m "feat(worker): share the completion cache between sink and session"`.
|
||||
|
||||
### Task 5: Executor bounded wait + StaSession env plumbing
|
||||
|
||||
**Classification:** high-risk (STA/pump semantics)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (touches the same files as 6's tests exercise)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs` (~lines 14-85 ctors, 447-497 write methods)
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs` (~line 205 executor construction)
|
||||
|
||||
**Step 1:** Executor — next to `DefaultReadBulkTimeout`:
|
||||
|
||||
```csharp
|
||||
/// <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 TimeSpan writeCompletionTimeout;
|
||||
```
|
||||
|
||||
Add `TimeSpan? writeCompletionTimeout = null` as a trailing optional parameter on the widest (4-arg) ctor; `this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout;`.
|
||||
|
||||
**Step 2:** In `ExecuteWriteSecured`, replace the tail (`session.WriteSecured(...); return CreateOkReply(command);`) with:
|
||||
|
||||
```csharp
|
||||
MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
|
||||
// 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).
|
||||
ulong completionBaseline = completionCache.CurrentVersion(
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle);
|
||||
|
||||
session.WriteSecured(
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle,
|
||||
writeSecuredCommand.CurrentUserId,
|
||||
writeSecuredCommand.VerifierUserId,
|
||||
variantConverter.ConvertToComValue(writeSecuredCommand.Value));
|
||||
|
||||
MxCommandReply reply = CreateOkReply(command);
|
||||
AwaitWriteCompletion(
|
||||
reply,
|
||||
completionCache,
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle,
|
||||
completionBaseline);
|
||||
return reply;
|
||||
```
|
||||
|
||||
Mirror in `ExecuteWriteSecured2`. Shared private helper:
|
||||
|
||||
```csharp
|
||||
/// <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);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Step 3:** `MxAccessStaSession` — add:
|
||||
|
||||
```csharp
|
||||
/// <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";
|
||||
|
||||
/// <summary>
|
||||
/// Bounded WriteSecured/WriteSecured2 completion wait handed to the
|
||||
/// command executor at StartAsync. Internal-settable as a test seam so
|
||||
/// Worker.Tests can shorten it without env-var plumbing.
|
||||
/// </summary>
|
||||
internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout();
|
||||
|
||||
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;
|
||||
}
|
||||
```
|
||||
|
||||
and pass `writeCompletionTimeout: WriteCompletionTimeout` when constructing `MxAccessCommandExecutor` in `StartAsync`. (net48: `Environment.GetEnvironmentVariable` returns `string` — keep nullable annotations consistent with the file.)
|
||||
|
||||
**Step 4:** Commit: `git commit -m "feat(worker): bounded pump-wait correlates OnWriteComplete onto secured-write replies"`.
|
||||
|
||||
### Task 6: Executor tests
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (depends on Tasks 2-5)
|
||||
|
||||
**Files:**
|
||||
- Test: `src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs`
|
||||
|
||||
**Step 1:** Test support inside the test class file:
|
||||
- New sink: `private sealed class CompletionCacheEventSink : IMxAccessEventSink, IWriteCompletionCacheProvider { public MxAccessWriteCompletionCache WriteCompletionCache { get; } = new MxAccessWriteCompletionCache(); public void Attach(object mxAccessComObject, string sessionId) { } public void Detach() { } }` (match the exact `IMxAccessEventSink` member list — read the interface first).
|
||||
- `FakeMxAccessComObject`: add `public Action OnWriteSecuredCallback { get; set; }` (nullable per file convention) invoked at the end of `WriteSecured` and `WriteSecured2`.
|
||||
|
||||
**Step 2:** Tests (all through `MxAccessStaSession.DispatchAsync`, constructing the session with the completion sink; set `session.WriteCompletionTimeout` before `StartAsync`):
|
||||
|
||||
- `DispatchAsync_WriteSecured_WhenCompletionArrivesDuringComCall_ReturnsStatuses` (fast-completion edge): wire `OnWriteSecuredCallback = () => sink.WriteCompletionCache.Record(82, 821, StatusRows(1))` (helper building a `RepeatedField<MxStatusProxy>` with a recognizable value); dispatch; assert `ProtocolStatus.Code == Ok`, `reply.Statuses.Count == 1`, row round-trips.
|
||||
- `DispatchAsync_WriteSecured_WhenCompletionArrivesWhileWaiting_ReturnsStatuses`: no fake callback; `WriteCompletionTimeout = TimeSpan.FromSeconds(10)`; start `Task<MxCommandReply> pending = session.DispatchAsync(...)`, then `sink.WriteCompletionCache.Record(...)` from the test thread after a short `Task.Delay(50)`; await; assert statuses present. (No `.ConfigureAwait(false)` in `[Fact]` bodies — xUnit1030 fails the Windows build.)
|
||||
- `DispatchAsync_WriteSecured_WhenNoCompletion_TimesOutWithEmptyStatusesAndOkProtocol`: `WriteCompletionTimeout = TimeSpan.FromMilliseconds(100)`; assert Ok + `reply.Statuses.Count == 0`.
|
||||
- `DispatchAsync_WriteSecured2_WhenCompletionArrivesDuringComCall_ReturnsStatuses` (mirror of the fast test).
|
||||
- `DispatchAsync_Write_DoesNotWaitForCompletion`: `WriteCompletionTimeout = TimeSpan.FromSeconds(30)`, plain `Write` command, no completion recorded; assert `await` completes within a 5 s guard (`Task.WhenAny` with `Task.Delay`) — proving plain writes never enter the wait.
|
||||
- Baseline test `DispatchAsync_WriteSecured_IgnoresStaleCompletionFromBeforeTheCall`: `Record` once BEFORE dispatch, `WriteCompletionTimeout = 100 ms`, no new completion → empty statuses (stale row not misattributed).
|
||||
|
||||
**Step 2b:** Confirm the two existing WriteSecured tests (`DispatchAsync_WriteSecured_ForwardsUserIds`, `..._WriteSecured2_...`) still pass unmodified — they use `NoopEventSink`, so the session falls back to a fresh cache, no completion ever arrives, and the default 1.5 s wait adds latency only; if that latency bothers the suite, switch them to the completion sink with `WriteCompletionTimeout = TimeSpan.Zero`.
|
||||
|
||||
**Step 3:** Commit: `git commit -m "test(worker): write-completion correlation executor coverage"`.
|
||||
|
||||
### Task 7: Gateway config option + launcher env var
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 2, Task 3, Task 4
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs`
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs` (~line 224 block)
|
||||
- Modify: `src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs` (~lines 18-22 consts, ~line 175 env block)
|
||||
- Test: `src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs`, the `GatewayOptionsValidator` test file (find via `grep -rl GatewayOptionsValidatorTests src/ZB.MOM.WW.MxGateway.Tests`)
|
||||
|
||||
**Step 1:** `WorkerOptions`:
|
||||
|
||||
```csharp
|
||||
/// <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;
|
||||
```
|
||||
|
||||
**Step 2:** Validator (>= 0, not the positive helper):
|
||||
|
||||
```csharp
|
||||
if (options.WriteCompletionWaitMilliseconds < 0)
|
||||
{
|
||||
builder.Add("MxGateway:Worker:WriteCompletionWaitMilliseconds must be greater than or equal to zero.");
|
||||
}
|
||||
```
|
||||
|
||||
**Step 3:** Launcher — const `public const string WorkerWriteCompletionWaitEnvironmentVariableName = "MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";` and in `CreateStartInfo` next to the pipe-connect env line:
|
||||
|
||||
```csharp
|
||||
startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
|
||||
_workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
```
|
||||
|
||||
**Step 4:** Tests — mirror the existing pipe-connect-timeout launcher env assertion and an existing validator negative test; add: launcher exports `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS=1500` by default / custom value when configured; validator rejects `-1`, accepts `0`.
|
||||
|
||||
**Step 5:** Run locally:
|
||||
|
||||
```bash
|
||||
dotnet test src/ZB.MOM.WW.MxGateway.Tests/ZB.MOM.WW.MxGateway.Tests.csproj --filter "FullyQualifiedName~WorkerProcessLauncherTests|FullyQualifiedName~GatewayOptionsValidator"
|
||||
```
|
||||
Expected: PASS.
|
||||
|
||||
**Step 6:** Commit: `git commit -m "feat(gateway): configurable worker write-completion wait (MxGateway:Worker:WriteCompletionWaitMilliseconds)"`.
|
||||
|
||||
### Task 8: Docs
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** none (write after code settles)
|
||||
|
||||
**Files:**
|
||||
- Modify: `docs/GatewayConfiguration.md` (Worker options table, ~line 113 area)
|
||||
- Modify: `gateway.md` (command surface / write semantics section)
|
||||
- Modify: `docs/DesignDecisions.md` (new decision entry)
|
||||
|
||||
Content: the option row (default 1500, 0 disables, consumer-budget pairing rule with OtOpcUa's Write resilience timeout); gateway.md note that WriteSecured/WriteSecured2 unary replies now carry correlated completion statuses (bounded wait, empty = unconfirmed, event stream unchanged); DesignDecisions entry summarizing the design doc (link it) including the best-effort per-(hItem) correlation caveat and why plain Write/Write2/bulk stay fire-and-forget. Follow `docs/style-guides/StyleGuide.md` (present tense, why not what).
|
||||
|
||||
Commit: `git commit -m "docs: write-completion correlation configuration and semantics"`.
|
||||
|
||||
### Task 9: Local + windev verification
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~10 min (mostly remote build time)
|
||||
**Parallelizable with:** none
|
||||
|
||||
**Step 1 (local):** `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` → succeeds; re-run the Task 7 filtered gateway tests.
|
||||
|
||||
**Step 2 (push):** `git push -u origin feat/write-completion-correlation`.
|
||||
|
||||
**Step 3 (windev):** Use the isolated `C:\build` worktree (NOT the Desktop checkout — dirty feature branch). Remote PS via base64 `-EncodedCommand`; no `$ErrorActionPreference='Stop'` around git. Sequence:
|
||||
|
||||
```
|
||||
git fetch origin && git checkout feat/write-completion-correlation && git pull
|
||||
dotnet build src/ZB.MOM.WW.MxGateway.Worker/ZB.MOM.WW.MxGateway.Worker.csproj -p:Platform=x86
|
||||
dotnet test src/ZB.MOM.WW.MxGateway.Worker.Tests/ZB.MOM.WW.MxGateway.Worker.Tests.csproj -p:Platform=x86 --filter "FullyQualifiedName~MxAccessWriteCompletionCacheTests|FullyQualifiedName~MxAccessBaseEventSinkTests|FullyQualifiedName~MxAccessCommandExecutorTests"
|
||||
```
|
||||
Expected: build clean (TreatWarningsAsErrors — watch xUnit1030), all filtered tests PASS. Fix-and-push iterations happen from the Mac; windev only builds/tests.
|
||||
|
||||
**Step 4:** Full worker test suite once green on the filter (`dotnet test ... -p:Platform=x86`, no filter) — one full pass before merge.
|
||||
|
||||
### Task 10: Review, merge, notify
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none
|
||||
|
||||
- Run a code review over the full branch diff (code-reviewer agent or /code-review) and address findings.
|
||||
- Merge: `git checkout main && git merge --no-ff feat/write-completion-correlation && git push origin main`.
|
||||
- Notify the OtOpcUa session (SendMessage) that the feature is on `main` and built/tested on windev; flag that the live ArchestrA leg needs the user's Windows deployment (wonder-app-vd03 / 10.100.0.48 redeploy is a separate user-approved step).
|
||||
- Surface to the user: deployed services (`MxAccessGw` on 10.100.0.48, wonder-app-vd03) do NOT pick this up until redeployed; OtOpcUa's end-to-end verification against a live gateway needs that redeploy.
|
||||
@@ -0,0 +1,17 @@
|
||||
{
|
||||
"planPath": "docs/plans/2026-08-09-write-completion-correlation.md",
|
||||
"tasks": [
|
||||
{"id": 0, "subject": "Task 0: Create feature branch", "status": "pending"},
|
||||
{"id": 1, "subject": "Task 1: Proto contract comments + regen", "status": "pending", "blockedBy": [0]},
|
||||
{"id": 2, "subject": "Task 2: MxAccessWriteCompletionCache + tests", "status": "pending", "blockedBy": [1]},
|
||||
{"id": 3, "subject": "Task 3: Sink records completions (+ provider seam)", "status": "pending", "blockedBy": [2]},
|
||||
{"id": 4, "subject": "Task 4: MxAccessSession plumbing", "status": "pending", "blockedBy": [3]},
|
||||
{"id": 5, "subject": "Task 5: Executor bounded wait + StaSession env plumbing", "status": "pending", "blockedBy": [4]},
|
||||
{"id": 6, "subject": "Task 6: Executor tests", "status": "pending", "blockedBy": [5]},
|
||||
{"id": 7, "subject": "Task 7: Gateway config option + launcher env var", "status": "pending", "blockedBy": [0]},
|
||||
{"id": 8, "subject": "Task 8: Docs", "status": "pending", "blockedBy": [6, 7]},
|
||||
{"id": 9, "subject": "Task 9: Local + windev verification", "status": "pending", "blockedBy": [8]},
|
||||
{"id": 10, "subject": "Task 10: Review, merge, notify", "status": "pending", "blockedBy": [9]}
|
||||
],
|
||||
"lastUpdated": "2026-08-09"
|
||||
}
|
||||
+17
@@ -431,6 +431,23 @@ Core commands:
|
||||
- `AuthenticateUser`
|
||||
- `ArchestrAUserToId`
|
||||
|
||||
**Secured-write completion correlation.** MXAccess writes are fire-and-forget
|
||||
at the toolkit level — the per-item outcome only exists in the later
|
||||
`OnWriteComplete` callback. For `WriteSecured` and `WriteSecured2` the worker
|
||||
therefore holds the unary reply for a bounded window
|
||||
(`MxGateway:Worker:WriteCompletionWaitMilliseconds`, default 1.5 s, `0`
|
||||
disables) and, when the matching callback arrives, copies its status rows onto
|
||||
`MxCommandReply.statuses` — the reply then proves the MXAccess commit, not just
|
||||
command acceptance. `protocol_status`/`hresult` keep describing acceptance
|
||||
only; a real MXAccess write failure surfaces in `statuses[0]`, and a reply with
|
||||
empty `statuses` means unconfirmed (the callback missed the window), never
|
||||
failed. The `OnWriteComplete` event still flows on the event stream unchanged.
|
||||
Correlation is best-effort per `(server_handle, item_handle)` — the callback
|
||||
carries no transaction id, so concurrent writes to the same item within the
|
||||
window can swap rows. Plain `Write`/`Write2` and the bulk write commands stay
|
||||
fire-and-forget: waiting there would add a device round-trip of latency to
|
||||
high-rate supervisory write loops.
|
||||
|
||||
Bulk variants (single gRPC round-trip carries the full list, the worker
|
||||
runs the per-item MXAccess calls sequentially on its STA, and the reply
|
||||
returns one result per requested entry — per-entry failures populate
|
||||
|
||||
@@ -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}")]
|
||||
public sealed partial class WriteSecuredCommand : pb::IMessage<WriteSecuredCommand>
|
||||
#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}")]
|
||||
public sealed partial class WriteSecured2Command : pb::IMessage<WriteSecured2Command>
|
||||
#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
|
||||
= 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>();
|
||||
/// <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.CodeDom.Compiler.GeneratedCode("protoc", null)]
|
||||
public pbc::RepeatedField<global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxStatusProxy> Statuses {
|
||||
|
||||
@@ -256,6 +256,9 @@ message Write2Command {
|
||||
int32 user_id = 5;
|
||||
}
|
||||
|
||||
// The unary reply's statuses field carries the correlated OnWriteComplete
|
||||
// outcome when it arrives within the worker's bounded wait — see
|
||||
// MxCommandReply.statuses.
|
||||
message WriteSecuredCommand {
|
||||
int32 server_handle = 1;
|
||||
int32 item_handle = 2;
|
||||
@@ -266,6 +269,9 @@ message WriteSecuredCommand {
|
||||
MxValue value = 5;
|
||||
}
|
||||
|
||||
// The unary reply's statuses field carries the correlated OnWriteComplete
|
||||
// outcome when it arrives within the worker's bounded wait — see
|
||||
// MxCommandReply.statuses.
|
||||
message WriteSecured2Command {
|
||||
int32 server_handle = 1;
|
||||
int32 item_handle = 2;
|
||||
@@ -525,6 +531,18 @@ message MxCommandReply {
|
||||
// transport failures.
|
||||
optional int32 hresult = 5;
|
||||
MxValue return_value = 6;
|
||||
// Correlated per-item outcome rows. For WRITE_SECURED / WRITE_SECURED2
|
||||
// replies the worker holds the reply for a bounded window (default 1.5 s,
|
||||
// MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for the matching
|
||||
// MXAccess OnWriteComplete callback and copies its status rows here, so
|
||||
// statuses[0] carries the real MXAccess commit outcome (success OR failure)
|
||||
// while protocol_status/hresult still describe command acceptance only.
|
||||
// Empty statuses on a write reply means the completion did not arrive
|
||||
// within the window — the write is unconfirmed, not failed. Correlation is
|
||||
// best-effort per (server_handle, item_handle): MXAccess's callback carries
|
||||
// no transaction id, so concurrent writes to the same item within the
|
||||
// window can swap rows. The OnWriteComplete event still flows on the event
|
||||
// stream unchanged. Other command kinds leave this field as before.
|
||||
repeated MxStatusProxy statuses = 7;
|
||||
string diagnostic_message = 8;
|
||||
|
||||
|
||||
@@ -225,6 +225,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
|
||||
options.PipeConnectAttemptTimeoutMilliseconds,
|
||||
"MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds must be greater than zero.",
|
||||
builder);
|
||||
if (options.WriteCompletionWaitMilliseconds < 0)
|
||||
{
|
||||
builder.Add("MxGateway:Worker:WriteCompletionWaitMilliseconds must be greater than or equal to zero.");
|
||||
}
|
||||
AddIfNotPositive(
|
||||
options.ShutdownTimeoutSeconds,
|
||||
"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>
|
||||
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>
|
||||
public int ShutdownTimeoutSeconds { get; init; } = 10;
|
||||
|
||||
|
||||
@@ -21,6 +21,14 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
|
||||
public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName =
|
||||
"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 IWorkerStartupProbe _startupProbe;
|
||||
private readonly GatewayMetrics _metrics;
|
||||
@@ -175,6 +183,8 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
|
||||
startInfo.Environment[WorkerNonceEnvironmentVariableName] = request.Nonce;
|
||||
startInfo.Environment[WorkerPipeConnectAttemptTimeoutEnvironmentVariableName] =
|
||||
_workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
|
||||
_workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
|
||||
commandLine = new WorkerProcessCommandLine(executablePath, arguments);
|
||||
|
||||
|
||||
@@ -41,6 +41,7 @@ public sealed class GatewayOptionsTests
|
||||
Assert.Equal(3, options.Worker.StartupProbeRetryAttempts);
|
||||
Assert.Equal(250, options.Worker.StartupProbeRetryDelayMilliseconds);
|
||||
Assert.Equal(2000, options.Worker.PipeConnectAttemptTimeoutMilliseconds);
|
||||
Assert.Equal(1500, options.Worker.WriteCompletionWaitMilliseconds);
|
||||
Assert.Equal(10, options.Worker.ShutdownTimeoutSeconds);
|
||||
Assert.Equal(5, options.Worker.HeartbeatIntervalSeconds);
|
||||
Assert.Equal(15, options.Worker.HeartbeatGraceSeconds);
|
||||
@@ -106,6 +107,7 @@ public sealed class GatewayOptionsTests
|
||||
[InlineData("MxGateway:Worker:ExecutablePath", "worker.dll", "MxGateway:Worker:ExecutablePath must point to a .exe file.")]
|
||||
[InlineData("MxGateway:Worker:StartupProbeRetryAttempts", "0", "MxGateway:Worker:StartupProbeRetryAttempts must be greater than zero.")]
|
||||
[InlineData("MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds", "0", "MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds must be greater than zero.")]
|
||||
[InlineData("MxGateway:Worker:WriteCompletionWaitMilliseconds", "-1", "MxGateway:Worker:WriteCompletionWaitMilliseconds must be greater than or equal to zero.")]
|
||||
[InlineData("MxGateway:Sessions:DefaultLeaseSeconds", "0", "MxGateway:Sessions:DefaultLeaseSeconds must be greater than zero.")]
|
||||
[InlineData("MxGateway:Sessions:LeaseSweepIntervalSeconds", "0", "MxGateway:Sessions:LeaseSweepIntervalSeconds must be greater than zero.")]
|
||||
[InlineData("MxGateway:Sessions:DetachGraceSeconds", "-1", "MxGateway:Sessions:DetachGraceSeconds must be zero or greater (0 disables detach-grace retention).")]
|
||||
|
||||
@@ -43,6 +43,10 @@ public sealed class WorkerProcessLauncherTests
|
||||
"2000",
|
||||
processFactory.LastStartInfo.Environment[
|
||||
WorkerProcessLauncher.WorkerPipeConnectAttemptTimeoutEnvironmentVariableName]);
|
||||
Assert.Equal(
|
||||
"1500",
|
||||
processFactory.LastStartInfo.Environment[
|
||||
WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]);
|
||||
Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal);
|
||||
Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal);
|
||||
Assert.False(pipeReservation.DisposeCalled);
|
||||
|
||||
@@ -110,6 +110,47 @@ public sealed class MxAccessBaseEventSinkTests
|
||||
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>
|
||||
/// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences.
|
||||
/// </summary>
|
||||
|
||||
@@ -897,6 +897,9 @@ public sealed class MxAccessCommandExecutorTests
|
||||
FakeMxAccessComObjectFactory factory = new(fakeComObject);
|
||||
using StaRuntime runtime = CreateRuntime();
|
||||
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
|
||||
// No completion source in this test — disable the bounded reply wait
|
||||
// so the forwarding assertions don't pay the default 1.5 s timeout.
|
||||
session.WriteCompletionTimeout = TimeSpan.Zero;
|
||||
await session.StartAsync(workerProcessId: 1234);
|
||||
|
||||
MxCommandReply reply = await session.DispatchAsync(CreateWriteSecuredCommand(
|
||||
@@ -919,6 +922,8 @@ public sealed class MxAccessCommandExecutorTests
|
||||
FakeMxAccessComObjectFactory factory = new(fakeComObject);
|
||||
using StaRuntime runtime = CreateRuntime();
|
||||
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
|
||||
// Same rationale as the WriteSecured forwarding test above.
|
||||
session.WriteCompletionTimeout = TimeSpan.Zero;
|
||||
await session.StartAsync(workerProcessId: 1234);
|
||||
DateTime timestamp = new(2026, 5, 19, 13, 30, 0, DateTimeKind.Utc);
|
||||
|
||||
@@ -934,6 +939,181 @@ public sealed class MxAccessCommandExecutorTests
|
||||
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);
|
||||
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();
|
||||
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));
|
||||
await Task.Delay(50);
|
||||
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);
|
||||
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>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
@@ -1775,6 +1955,28 @@ public sealed class MxAccessCommandExecutorTests
|
||||
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 readonly int registerHandle;
|
||||
@@ -1790,6 +1992,14 @@ public sealed class MxAccessCommandExecutorTests
|
||||
private readonly IReadOnlyDictionary<int, Exception> writeExceptionByItemHandle;
|
||||
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>
|
||||
/// <param name="registerHandle">Return value for Register method.</param>
|
||||
/// <param name="addItemHandle">Return value for AddItem method.</param>
|
||||
@@ -2100,6 +2310,7 @@ public sealed class MxAccessCommandExecutorTests
|
||||
WriteValue = value;
|
||||
WriteThreadId = Environment.CurrentManagedThreadId;
|
||||
ThrowIfWriteFailureConfigured(itemHandle);
|
||||
OnWriteSecuredCallback?.Invoke();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -2120,6 +2331,7 @@ public sealed class MxAccessCommandExecutorTests
|
||||
WriteTimestamp = timestamp;
|
||||
WriteThreadId = Environment.CurrentManagedThreadId;
|
||||
ThrowIfWriteFailureConfigured(itemHandle);
|
||||
OnWriteSecuredCallback?.Invoke();
|
||||
}
|
||||
|
||||
private void ThrowIfWriteFailureConfigured(int itemHandle)
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <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 MxAccessEventQueue eventQueue;
|
||||
private readonly MxAccessValueCache valueCache;
|
||||
private readonly MxAccessWriteCompletionCache writeCompletionCache;
|
||||
private LMXProxyServerClass? server;
|
||||
private string sessionId = string.Empty;
|
||||
|
||||
@@ -50,10 +51,32 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
|
||||
MxAccessEventQueue eventQueue,
|
||||
MxAccessEventMapper eventMapper,
|
||||
MxAccessValueCache valueCache)
|
||||
: this(eventQueue, eventMapper, valueCache, new MxAccessWriteCompletionCache())
|
||||
{
|
||||
}
|
||||
|
||||
/// <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.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
|
||||
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
|
||||
this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -62,6 +85,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
|
||||
/// </summary>
|
||||
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 />
|
||||
public void Attach(
|
||||
object mxAccessComObject,
|
||||
@@ -143,11 +174,18 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
|
||||
ref MXSTATUS_PROXY[] pVars)
|
||||
{
|
||||
MXSTATUS_PROXY[] statuses = pVars;
|
||||
EnqueueEvent(() => eventMapper.CreateOnWriteComplete(
|
||||
sessionId,
|
||||
hLMXServerHandle,
|
||||
phItemHandle,
|
||||
statuses));
|
||||
// Record the completion for the write executor's bounded reply wait
|
||||
// only after the event has cleared the queue (same post-publish rule
|
||||
// as the OnDataChange value cache) — an overflow faults the session,
|
||||
// so a dropped event never leaves a "fresher" completion behind than
|
||||
// what shipped to the gateway.
|
||||
EnqueueEvent(
|
||||
() => eventMapper.CreateOnWriteComplete(
|
||||
sessionId,
|
||||
hLMXServerHandle,
|
||||
phItemHandle,
|
||||
statuses),
|
||||
mxEvent => writeCompletionCache.Record(hLMXServerHandle, phItemHandle, mxEvent.Statuses));
|
||||
}
|
||||
|
||||
/// <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>
|
||||
internal static readonly TimeSpan DefaultReadBulkTimeout = TimeSpan.FromMilliseconds(1000);
|
||||
|
||||
/// <summary>
|
||||
/// Default bounded wait for the OnWriteComplete callback after a
|
||||
/// WriteSecured/WriteSecured2 COM call. 1.5 s keeps the unary reply
|
||||
/// inside the OtOpcUa driver's 2 s Tier A write-resilience budget (a
|
||||
/// longer gateway wait must raise that consumer timeout in step) while
|
||||
/// covering the common fast-commit case; on expiry the reply returns
|
||||
/// with empty statuses — unconfirmed, not failed.
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan DefaultWriteCompletionTimeout = TimeSpan.FromMilliseconds(1500);
|
||||
|
||||
private readonly MxAccessSession session;
|
||||
private readonly VariantConverter variantConverter;
|
||||
private readonly MxStatusProxyConverter statusProxyConverter;
|
||||
private readonly IAlarmCommandHandler? alarmCommandHandler;
|
||||
private readonly Action pumpStep;
|
||||
private readonly TimeSpan writeCompletionTimeout;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a command executor with an MXAccess session.
|
||||
@@ -71,17 +82,26 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
/// <param name="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
|
||||
/// <param name="alarmCommandHandler">Optional handler for alarm-side commands.</param>
|
||||
/// <param name="pumpStep">Action to pump Windows messages, or null for tests.</param>
|
||||
/// <param name="writeCompletionTimeout">
|
||||
/// Bounded wait for the OnWriteComplete callback after a
|
||||
/// WriteSecured/WriteSecured2 COM call, or null for
|
||||
/// <see cref="DefaultWriteCompletionTimeout"/>. Zero (or negative)
|
||||
/// disables the wait entirely — replies keep the pure fire-and-forget
|
||||
/// shape.
|
||||
/// </param>
|
||||
public MxAccessCommandExecutor(
|
||||
MxAccessSession session,
|
||||
VariantConverter variantConverter,
|
||||
IAlarmCommandHandler? alarmCommandHandler,
|
||||
Action? pumpStep)
|
||||
Action? pumpStep,
|
||||
TimeSpan? writeCompletionTimeout = null)
|
||||
{
|
||||
this.session = session ?? throw new ArgumentNullException(nameof(session));
|
||||
this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter));
|
||||
this.statusProxyConverter = new MxStatusProxyConverter();
|
||||
this.alarmCommandHandler = alarmCommandHandler;
|
||||
this.pumpStep = pumpStep ?? (static () => { });
|
||||
this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -457,6 +477,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
return CreateInvalidRequestReply(command, "WriteSecured command value is required.");
|
||||
}
|
||||
|
||||
// Baseline BEFORE the COM call: a completion that dispatches during or
|
||||
// immediately after WriteSecured bumps the version past this snapshot,
|
||||
// so a fast commit still correlates (no missed-callback window).
|
||||
MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
|
||||
ulong completionBaseline = completionCache.CurrentVersion(
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle);
|
||||
|
||||
session.WriteSecured(
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle,
|
||||
@@ -464,7 +492,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
writeSecuredCommand.VerifierUserId,
|
||||
variantConverter.ConvertToComValue(writeSecuredCommand.Value));
|
||||
|
||||
return CreateOkReply(command);
|
||||
MxCommandReply reply = CreateOkReply(command);
|
||||
AwaitWriteCompletion(
|
||||
reply,
|
||||
completionCache,
|
||||
writeSecuredCommand.ServerHandle,
|
||||
writeSecuredCommand.ItemHandle,
|
||||
completionBaseline);
|
||||
return reply;
|
||||
}
|
||||
|
||||
private MxCommandReply ExecuteWriteSecured2(StaCommand command)
|
||||
@@ -485,6 +520,12 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
return CreateInvalidRequestReply(command, "WriteSecured2 command timestamp value is required.");
|
||||
}
|
||||
|
||||
// Same pre-call baseline rule as ExecuteWriteSecured.
|
||||
MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
|
||||
ulong completionBaseline = completionCache.CurrentVersion(
|
||||
writeSecured2Command.ServerHandle,
|
||||
writeSecured2Command.ItemHandle);
|
||||
|
||||
session.WriteSecured2(
|
||||
writeSecured2Command.ServerHandle,
|
||||
writeSecured2Command.ItemHandle,
|
||||
@@ -493,7 +534,14 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
variantConverter.ConvertToComValue(writeSecured2Command.Value),
|
||||
variantConverter.ConvertToComValue(writeSecured2Command.TimestampValue));
|
||||
|
||||
return CreateOkReply(command);
|
||||
MxCommandReply reply = CreateOkReply(command);
|
||||
AwaitWriteCompletion(
|
||||
reply,
|
||||
completionCache,
|
||||
writeSecured2Command.ServerHandle,
|
||||
writeSecured2Command.ItemHandle,
|
||||
completionBaseline);
|
||||
return reply;
|
||||
}
|
||||
|
||||
private MxCommandReply ExecuteAddItemBulk(StaCommand command)
|
||||
@@ -897,6 +945,39 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Bounded pump-wait for the OnWriteComplete row matching a
|
||||
/// WriteSecured/WriteSecured2 call, copied onto the reply when it
|
||||
/// arrives in time. The executor holds the STA thread but pumps
|
||||
/// Windows messages each poll (ReadBulk precedent) so the COM callback
|
||||
/// can dispatch re-entrantly; on expiry the reply keeps its empty
|
||||
/// statuses — the consumer's unconfirmed path, never a synthesized
|
||||
/// failure. Protocol status/hresult stay acceptance-only either way.
|
||||
/// </summary>
|
||||
private void AwaitWriteCompletion(
|
||||
MxCommandReply reply,
|
||||
MxAccessWriteCompletionCache completionCache,
|
||||
int serverHandle,
|
||||
int itemHandle,
|
||||
ulong completionBaseline)
|
||||
{
|
||||
if (writeCompletionTimeout <= TimeSpan.Zero)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (completionCache.TryWaitForCompletion(
|
||||
serverHandle,
|
||||
itemHandle,
|
||||
completionBaseline,
|
||||
DateTime.UtcNow + writeCompletionTimeout,
|
||||
pumpStep,
|
||||
out Google.Protobuf.Collections.RepeatedField<MxStatusProxy> statuses))
|
||||
{
|
||||
reply.Statuses.Add(statuses);
|
||||
}
|
||||
}
|
||||
|
||||
private static MxCommandReply CreateAlarmFailureReply(StaCommand command, Exception exception)
|
||||
{
|
||||
return new MxCommandReply
|
||||
|
||||
@@ -13,6 +13,7 @@ public sealed class MxAccessSession : IDisposable
|
||||
private readonly IMxAccessEventSink eventSink;
|
||||
private readonly MxAccessHandleRegistry handleRegistry;
|
||||
private readonly MxAccessValueCache valueCache;
|
||||
private readonly MxAccessWriteCompletionCache writeCompletionCache;
|
||||
private bool disposed;
|
||||
|
||||
private MxAccessSession(
|
||||
@@ -21,6 +22,7 @@ public sealed class MxAccessSession : IDisposable
|
||||
IMxAccessEventSink eventSink,
|
||||
MxAccessHandleRegistry handleRegistry,
|
||||
MxAccessValueCache valueCache,
|
||||
MxAccessWriteCompletionCache writeCompletionCache,
|
||||
int creationThreadId)
|
||||
{
|
||||
this.mxAccessComObject = mxAccessComObject ?? throw new ArgumentNullException(nameof(mxAccessComObject));
|
||||
@@ -28,6 +30,7 @@ public sealed class MxAccessSession : IDisposable
|
||||
this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
|
||||
this.handleRegistry = handleRegistry ?? throw new ArgumentNullException(nameof(handleRegistry));
|
||||
this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
|
||||
this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
|
||||
CreationThreadId = creationThreadId;
|
||||
}
|
||||
|
||||
@@ -45,6 +48,14 @@ public sealed class MxAccessSession : IDisposable
|
||||
/// </summary>
|
||||
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>
|
||||
/// <param name="workerProcessId">Process ID of the worker.</param>
|
||||
/// <returns>The populated <see cref="WorkerReady"/> message.</returns>
|
||||
@@ -149,12 +160,22 @@ public sealed class MxAccessSession : IDisposable
|
||||
? baseSink.ValueCache
|
||||
: new MxAccessValueCache();
|
||||
|
||||
// Share the sink's completion cache the same way (the production
|
||||
// sink and completion-aware test sinks implement the provider
|
||||
// seam); fall back to a fresh cache for other fakes — the write
|
||||
// executor then simply never observes a completion and replies
|
||||
// unconfirmed.
|
||||
MxAccessWriteCompletionCache writeCompletionCache = eventSink is IWriteCompletionCacheProvider provider
|
||||
? provider.WriteCompletionCache
|
||||
: new MxAccessWriteCompletionCache();
|
||||
|
||||
return new MxAccessSession(
|
||||
mxAccessComObject,
|
||||
new MxAccessComServer(mxAccessComObject),
|
||||
eventSink,
|
||||
new MxAccessHandleRegistry(),
|
||||
valueCache,
|
||||
writeCompletionCache,
|
||||
Environment.CurrentManagedThreadId);
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
@@ -11,6 +11,14 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
|
||||
|
||||
public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
{
|
||||
/// <summary>
|
||||
/// Environment variable the gateway's WorkerProcessLauncher sets from
|
||||
/// MxGateway:Worker:WriteCompletionWaitMilliseconds. 0 disables the
|
||||
/// write-completion wait (pure fire-and-forget replies).
|
||||
/// </summary>
|
||||
internal const string WriteCompletionWaitEnvironmentVariableName =
|
||||
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
|
||||
|
||||
private static readonly TimeSpan AlarmPollInterval = TimeSpan.FromMilliseconds(500);
|
||||
|
||||
private readonly IMxAccessComObjectFactory factory;
|
||||
@@ -157,6 +165,32 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
/// </summary>
|
||||
public MxAccessEventQueue EventQueue => eventQueue;
|
||||
|
||||
/// <summary>
|
||||
/// Bounded WriteSecured/WriteSecured2 completion wait handed to the
|
||||
/// command executor at <see cref="StartAsync(string, int, CancellationToken)"/>.
|
||||
/// Internal-settable as a test seam so Worker.Tests can shorten it
|
||||
/// without env-var plumbing.
|
||||
/// </summary>
|
||||
internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout();
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the write-completion wait from the launcher-provided
|
||||
/// environment variable; a missing or invalid value falls back to
|
||||
/// <see cref="MxAccessCommandExecutor.DefaultWriteCompletionTimeout"/>.
|
||||
/// </summary>
|
||||
internal static TimeSpan ResolveWriteCompletionTimeout()
|
||||
{
|
||||
string? value = Environment.GetEnvironmentVariable(WriteCompletionWaitEnvironmentVariableName);
|
||||
return int.TryParse(
|
||||
value,
|
||||
System.Globalization.NumberStyles.Integer,
|
||||
System.Globalization.CultureInfo.InvariantCulture,
|
||||
out int milliseconds)
|
||||
&& milliseconds >= 0
|
||||
? TimeSpan.FromMilliseconds(milliseconds)
|
||||
: MxAccessCommandExecutor.DefaultWriteCompletionTimeout;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the MXAccess COM session asynchronously.
|
||||
/// </summary>
|
||||
@@ -208,12 +242,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
session,
|
||||
new VariantConverter(),
|
||||
alarmCommandHandler,
|
||||
// ReadBulk needs to pump Windows messages while it waits
|
||||
// for the first OnDataChange callback so the inbound COM
|
||||
// event can dispatch on this same STA thread. The pump
|
||||
// step closes over staRuntime so it always pumps the
|
||||
// pump tied to the apartment that owns this session.
|
||||
pumpStep: () => staRuntime.PumpPendingMessages()));
|
||||
// ReadBulk and the write-completion wait need to pump
|
||||
// Windows messages while they wait for the inbound COM
|
||||
// callback (OnDataChange / OnWriteComplete) so it can
|
||||
// dispatch on this same STA thread. The pump step
|
||||
// closes over staRuntime so it always pumps the pump
|
||||
// tied to the apartment that owns this session.
|
||||
pumpStep: () => staRuntime.PumpPendingMessages(),
|
||||
writeCompletionTimeout: WriteCompletionTimeout));
|
||||
|
||||
return session.CreateWorkerReady(workerProcessId);
|
||||
},
|
||||
|
||||
@@ -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; }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user