dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(galaxy): consume the gateway's correlated write statuses; close archreview 06/S-1

Browse Source 
The mxaccessgw OnWriteComplete correlation is live (server b948e69, worker
53f69cd — all four unary write kinds), so every write reply's statuses[0] now
carries the real COM commit outcome. Live-gating the cutover on windev found
and fixed three defects:

- StatusCodeMap.FromMxStatus truncated the correlated failure row's MxStatus
  detail code to a byte (1007 & 0xFF = 239 -> OPC DA Good band), mapping a
  REFUSED write to Good. New invariant: a row that is not IsSuccess() never
  maps Good — details > 0xFF (and contradictory Good-band substatuses) route
  by category (SECURITY_ERROR -> BadUserAccessDenied, named const covered by
  StatusCodeParityTests; COMMUNICATION/CONFIGURATION keep their codes).

- The borrow-from-subscription-registry write optimization could never commit:
  a subscription handle is plain-advised, and MXAccess refuses both the
  no-login write through it (SECURITY_ERROR detail 1008, live-captured) and
  the supervisory advise upgrade (E_INVALIDARG) — one advise per HANDLE, not
  per attribute. Removed: writer subscribedHandleSource ctor param + borrow
  branch, GalaxyDriver wiring, SubscriptionRegistry.TryResolveItemHandle +
  its forward index and test file. The writer always AddItems its own
  supervisory-advised handle, which commits fine while the same attribute is
  subscribed (live-proven; pinned by the new write-while-subscribed smoke).

- (gateway-side, fixed there) the correlation initially covered WriteSecured
  only, while the dominant no-login path emits plain Write.

Live gate, all legs PASSED against windev: happy write Good off a confirmed
status row with galaxy.writes.unconfirmed + advise_failed both FLAT; read-only
refusal surfaces Bad-family (the #5 node-revert input Galaxy could never
produce before); WriteCompletionWaitMilliseconds=0 control -> provisional Good
+ metered unconfirmed (override removed after, restoration re-verified). New
env-gated smokes: Live_correlated_write_statuses_confirm_commit_and_surface_refusal,
Live_write_to_subscribed_tag_commits_via_own_handle (replaces the borrow
smoke), GatewayGalaxyWriteMeterLiveTests (run filtered — meters are
process-wide).
This commit is contained in:
Joseph Doherty
2026-08-09 20:11:55 -04:00
parent 1038c0e23c
commit 20a816dbf4
11 changed files with 337 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+6 -4
View File
@@ -378,11 +378,13 @@ public sealed class GalaxyDriver
_subscriber = new TracedGalaxySubscriber(
new GatewayGalaxySubscriber(_ownedMxSession, _logger), _options.MxAccess.ClientName);
_dataWriter = new TracedGalaxyDataWriter(
// Let the writer borrow live MXAccess item handles the subscription registry already
// holds, so the first write to an already-subscribed tag skips a redundant AddItem.
// The writer always AddItems its own handle — a subscription's handle is plain-advised,
// and MXAccess both refuses a no-login write through it (SECURITY_ERROR 1008) and refuses
// upgrading its advise mode (E_INVALIDARG), so the former borrow-from-registry
// optimization could never commit a write. The advise limit is per handle, not per
// attribute: the writer's own supervisory-advised handle coexists with the subscription's.
new GatewayGalaxyDataWriter(
_ownedMxSession, _options.MxAccess.WriteUserId, _logger,
_subscriptions.TryResolveItemHandle),
_ownedMxSession, _options.MxAccess.WriteUserId, _logger),
_options.MxAccess.ClientName);
_supervisor = new ReconnectSupervisor(
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs
+26 -51
View File
@@ -18,19 +18,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
/// <remarks>
/// Item handle cache survives across writes — repeated writes to the same tag avoid
/// re-AddItem. Per-tag failures are isolated: one bad write doesn't fail the batch.
/// When a <c>subscribedHandleSource</c> delegate is supplied, the first write to an
/// already-subscribed tag borrows the live handle from the subscription registry and
/// skips the AddItem round-trip entirely. Borrowed handles are intentionally NOT stored
/// in <c>_itemHandles</c> — the subscription registry owns their lifecycle (including
/// reconnect rebind), so each write re-borrows the fresh handle and no stale-cache
/// window is introduced.
/// The writer always owns its handles: it never borrows a subscription's live handle,
/// because a subscription handle is plain-advised and MXAccess enforces one advise per
/// handle — a no-login write through a plain-advised handle is refused (SECURITY_ERROR
/// detail 1008) and re-advising it supervisory is rejected (E_INVALIDARG), both
/// live-captured on the correlated write replies (06/S-1). The advise limit is per
/// handle, not per attribute, so the writer's own supervisory-advised handle coexists
/// with the subscription's plain-advised one on the same attribute.
/// </remarks>
public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
{
private readonly GalaxyMxSession _session;
private readonly int _writeUserId;
private readonly ILogger _logger;
private readonly Func<string, int?>? _subscribedHandleSource;
private readonly ConcurrentDictionary<string, int> _itemHandles =
new(StringComparer.OrdinalIgnoreCase);
// Item handles we've already AdviseSupervisory'd this session — supervisory advise is
@@ -41,7 +41,9 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
// Write failure-visibility meters (archreview 06/S-1). Reuse the pump's meter name so the single host
// listener catches both. advise_failed = writes short-circuited to Bad because AdviseSupervisory failed
// (the value could not have committed); unconfirmed = writes reported Good off an EMPTY gateway statuses
// array (command accepted, COM commit unconfirmed pending the gateway WriteComplete correlation follow-up).
// array. A correlated gateway (mxaccessgw ≥ 53f69cd rolls the worker's OnWriteComplete into every unary
// write reply's statuses[0]) keeps this counter FLAT — an increment means the gateway is pre-correlation
// or the completion wait timed out (WriteCompletionWaitMilliseconds), so the commit is unconfirmed.
private static readonly Meter WriterMeter = new(EventPump.MeterName);
private static readonly Counter<long> WriteAdviseFailed =
WriterMeter.CreateCounter<long>("galaxy.writes.advise_failed", unit: "{write}",
@@ -54,21 +56,11 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
/// <param name="session">The MXAccess gateway session.</param>
/// <param name="writeUserId">The user ID for write operations.</param>
/// <param name="logger">Optional logger for tracing.</param>
/// <param name="subscribedHandleSource">
/// Optional delegate that resolves a live MXAccess item handle from the subscription
/// registry for a given tag full reference. When the delegate returns a positive handle
/// the writer borrows it and skips the AddItem gateway round-trip. A return value of
/// <c>null</c>, 0, or negative is treated as "not available" and the writer falls back
/// to its own AddItem. Borrowed handles are NOT cached in this writer — the subscription
/// registry owns their lifecycle.
/// </param>
public GatewayGalaxyDataWriter(GalaxyMxSession session, int writeUserId, ILogger? logger = null,
Func<string, int?>? subscribedHandleSource = null)
public GatewayGalaxyDataWriter(GalaxyMxSession session, int writeUserId, ILogger? logger = null)
{
_session = session ?? throw new ArgumentNullException(nameof(session));
_writeUserId = writeUserId;
_logger = logger ?? NullLogger.Instance;
_subscribedHandleSource = subscribedHandleSource;
}
/// <inheritdoc />
@@ -94,7 +86,7 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
internal int CachedSupervisedHandleCount => _supervisedHandles.Count;
/// <summary>Count of real gateway <c>AddItem</c> round-trips this writer has issued. Stays zero
/// when every handle was served from cache or borrowed from the subscription registry. Test seam.</summary>
/// when every handle was served from the cache. Test seam.</summary>
internal int AddItemCallCount => Volatile.Read(ref _addItemCallCount);
/// <summary>
@@ -111,23 +103,6 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
if (supervised) _supervisedHandles.TryAdd(itemHandle, 0);
}
/// <summary>
/// Resolve an item handle WITHOUT touching the gateway: a handle this writer already AddItem'd
/// (<c>_itemHandles</c>), else a live subscription handle borrowed via
/// <c>_subscribedHandleSource</c>. Returns null when neither is available (the caller must
/// AddItem). A BORROWED handle is intentionally NOT stored in <c>_itemHandles</c> — the
/// subscription registry owns the borrowed handle's lifecycle (including reconnect rebind),
/// so the next write re-borrows the fresh handle and no stale-cache window is introduced.
/// </summary>
/// <param name="fullRef">The dotted tag full reference.</param>
/// <returns>A usable item handle, or null when an AddItem is required.</returns>
internal int? TryResolveCachedOrBorrowed(string fullRef)
{
if (_itemHandles.TryGetValue(fullRef, out var existing)) return existing;
if (_subscribedHandleSource?.Invoke(fullRef) is int borrowed && borrowed > 0) return borrowed;
return null;
}
/// <inheritdoc />
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<WriteRequest> writes,
@@ -190,13 +165,12 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
{
// A raw Write runs with NO user login (WriteUserId is typically 0), so MXAccess
// only COMMITS the value when the item is advised in SUPERVISORY mode. Without it
// the gateway's Write call doesn't throw (reply looks OK) but the value never
// reaches the galaxy — a knowingly-lost write. FAIL-CLOSED (archreview 06/S-1): when
// the write is refused by the automation object (SECURITY_ERROR detail 1008 on the
// correlated reply — live-captured). FAIL-CLOSED (archreview 06/S-1): when
// AdviseSupervisory fails we do NOT issue the raw Write; we return Bad so the
// #5 write-outcome self-correction reverts the phantom-Good node instead of leaving it.
// The real fix is a gateway-side WriteComplete correlation (mxaccessgw backlog) so the
// unary reply's Statuses carry the actual COM commit outcome; until then advise-status +
// the galaxy.writes.* meters are the honest signal.
// A correlated gateway (mxaccessgw ≥ 53f69cd) would also surface the refusal in the
// reply's statuses[0], but the short-circuit avoids a write we know cannot commit.
if (!await EnsureSupervisoryAdvisedAsync(ops, itemHandle, ct).ConfigureAwait(false))
{
WriteAdviseFailed.Add(1);
@@ -229,7 +203,7 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
private async Task<int> EnsureItemHandleAsync(
IMxWriteOps ops, string fullRef, CancellationToken ct)
{
if (TryResolveCachedOrBorrowed(fullRef) is int resolved) return resolved;
if (_itemHandles.TryGetValue(fullRef, out var existing)) return existing;
var handle = await ops.AddItemAsync(fullRef, ct).ConfigureAwait(false);
Interlocked.Increment(ref _addItemCallCount);
_itemHandles[fullRef] = handle;
@@ -323,12 +297,13 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
/// <summary>
/// Translate a gateway <see cref="MxCommandReply"/> into an OPC UA
/// <see cref="WriteResult"/>. Honours the protocol-level Status field first
/// (transport / dispatch failures), then the first MXAccess status row.
/// <para>An EMPTY statuses array is treated as <c>Good</c> but is PROVISIONAL: the reply proves
/// worker-side command acceptance, not the COM-side commit (the gateway does not yet correlate a
/// WriteComplete row onto the unary reply — mxaccessgw backlog). Each such reply increments
/// <c>galaxy.writes.unconfirmed</c> so the unconfirmed-write rate is operator-visible today
/// (archreview 06/S-1).</para>
/// (transport / dispatch failures), then the first MXAccess status row — which, against a
/// correlated gateway (mxaccessgw ≥ 53f69cd), is the worker's <c>OnWriteComplete</c> outcome:
/// the actual COM-side commit verdict, including refusals (archreview 06/S-1, closed).
/// <para>An EMPTY statuses array is treated as <c>Good</c> but is PROVISIONAL: it means the
/// gateway's bounded completion wait timed out (or the gateway pre-dates the correlation), so
/// the reply proves worker-side command acceptance only. Each such reply increments
/// <c>galaxy.writes.unconfirmed</c> so the unconfirmed-write rate stays operator-visible.</para>
/// </summary>
private WriteResult TranslateReply(MxCommandReply reply, string fullRef)
{
@@ -350,8 +325,8 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
return new WriteResult(StatusCodeMap.FromMxStatus(status, _logger));
}
// Empty statuses ⇒ provisional Good. Meter it so the unconfirmed-write rate is visible until the
// gateway-side WriteComplete correlation lands (archreview 06/S-1 cross-repo follow-up).
// Empty statuses ⇒ the gateway's completion wait timed out (or the gateway pre-dates the
// 53f69cd correlation) ⇒ provisional Good. Meter it so the unconfirmed-write rate is visible.
WriteUnconfirmed.Add(1);
return new WriteResult(StatusCodeMap.Good);
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs
+38 -4
View File
@@ -52,6 +52,7 @@ internal static class StatusCodeMap
public const uint BadOutOfService = 0x808D0000u;
public const uint BadWaitingForInitialData = 0x80320000u;
public const uint BadInternalError = 0x80020000u;
public const uint BadUserAccessDenied = 0x801F0000u;
/// <summary>
/// Fills a still-pending read when the caller's token fires before the gateway answers. Named here
@@ -111,10 +112,21 @@ internal static class StatusCodeMap
if (status is null) return Good;
if (status.IsSuccess()) return Good;
// Detail field carries the substatus when the worker translated MX-style codes;
// when zero, infer from detected_by.
var detail = (byte)(status.Detail & 0xFF);
if (detail != 0) return FromQualityByte(detail, logger);
// A detail in [1, 255] is an OPC DA quality substatus (read/data-change rows). Anything
// larger is an MXAccess MxStatus detail code — the write-completion rows the gateway
// correlates onto write replies carry these (e.g. 1007 operational, 1008 security) — and
// truncating one into the quality table maps a REFUSED WRITE to Good whenever the low byte
// lands in the Good band (1007 → 239 ≥ 192). Route those on the status category instead.
var detail = status.Detail;
if (detail is > 0 and <= 0xFF)
{
var mapped = FromQualityByte((byte)detail, logger);
// Invariant: a row that is not IsSuccess() must never surface as Good — a
// contradictory Good-band substatus on a failure row falls back to the category.
if ((mapped >> 30) != 0u) return mapped;
return FromCategory(status, logger);
}
if (detail > 0xFF) return FromCategory(status, logger);
// detected_by != Mxaccess (raw_detected_by != the MXAccess source enum) implies
// the failure happened pre-call (gateway, worker, transport) — surface as a
@@ -124,6 +136,28 @@ internal static class StatusCodeMap
return Bad;
}
/// <summary>
/// Map a failure row by its <see cref="MxStatusCategory"/> — used when the detail field is
/// an MxStatus detail code rather than an OPC DA quality byte (or contradicts the failure).
/// Category-level only: the MxStatus detail vocabulary (1007, 1008, …) is not mapped
/// per-code yet, so the log line carries the raw detail for field capture.
/// </summary>
private static uint FromCategory(MxStatusProxy status, ILogger? logger)
{
var code = status.Category switch
{
MxStatusCategory.SecurityError => BadUserAccessDenied,
MxStatusCategory.CommunicationError => BadCommunicationError,
MxStatusCategory.ConfigurationError => BadConfigurationError,
_ => Bad,
};
logger?.LogWarning(
"MxStatus failure row category {Category} detail {Detail} mapped to 0x{Code:X8} at category " +
"level. Field capture welcome — extend StatusCodeMap.FromCategory with the detail code.",
status.Category, status.Detail, code);
return code;
}
/// <summary>
/// Convert an OPC UA status-code uint back to the OPC DA quality category
/// byte — Good=192, Uncertain=64, Bad=0 — by extracting the top-two bits of the
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
-46
View File
@@ -20,12 +20,6 @@ internal sealed class SubscriptionRegistry
{
private readonly ConcurrentDictionary<long, SubscriptionEntry> _bySubscriptionId = new();
private readonly ConcurrentDictionary<int, ImmutableHashSet<long>> _subscribersByItemHandle = new();
// Forward index for the Galaxy writer: fullRef (case-insensitive) → live item handle.
// Maintained in lock-step with _subscribersByItemHandle; entries are cleaned up when
// the last subscriber for a handle is removed, and TryResolveItemHandle guards against
// stale entries by cross-checking _subscribersByItemHandle at read time.
private readonly ConcurrentDictionary<string, int> _itemHandleByFullRef =
new(StringComparer.OrdinalIgnoreCase);
private long _nextSubscriptionId;
/// <summary>Gets the number of tracked subscriptions.</summary>
@@ -55,7 +49,6 @@ internal sealed class SubscriptionRegistry
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
_itemHandleByFullRef[binding.FullReference] = binding.ItemHandle;
}
}
@@ -77,10 +70,6 @@ internal sealed class SubscriptionRegistry
if (remaining.IsEmpty)
{
_subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
// Clean up the forward index only when this handle is still the current
// mapping — a concurrent re-add for the same ref must not be clobbered.
if (_itemHandleByFullRef.TryGetValue(binding.FullReference, out var fwd) && fwd == binding.ItemHandle)
_itemHandleByFullRef.TryRemove(binding.FullReference, out _);
}
else _subscribersByItemHandle[binding.ItemHandle] = remaining;
}
@@ -110,36 +99,6 @@ internal sealed class SubscriptionRegistry
return result;
}
/// <summary>
/// Resolve the live MXAccess item handle a current subscription holds for <paramref name="fullRef"/>,
/// or null when no live subscription covers it. The Galaxy writer borrows this handle to skip a
/// redundant AddItem round-trip on the first write to an already-subscribed tag.
/// </summary>
/// <remarks>
/// The <c>_itemHandleByFullRef</c> forward map is only a hint — resolution is authoritative:
/// it confirms a <em>live</em> subscription actually binds this <paramref name="fullRef"/> to
/// the candidate handle (via the per-entry reverse index), so a stale forward entry can never
/// hand the writer a dead handle, nor a handle that now belongs to a different tag. This keeps
/// a tag resolvable while ANY subscription still binds it (the legitimate
/// multiple-driver-subscriptions-per-tag case) yet returns null the moment the last one drops.
/// </remarks>
/// <param name="fullRef">The dotted tag full reference (e.g. <c>TestMachine_002.TestFloat</c>).</param>
/// <returns>The live item handle, or null when none is currently subscribed.</returns>
public int? TryResolveItemHandle(string fullRef)
{
if (fullRef is null) return null;
if (!_itemHandleByFullRef.TryGetValue(fullRef, out var handle)) return null;
if (!_subscribersByItemHandle.TryGetValue(handle, out var subs)) return null; // dead handle
// Confirm a live subscription genuinely binds fullRef -> handle (not a stale forward entry
// pointing at a handle the gateway has since associated with another tag).
foreach (var subId in subs)
if (_bySubscriptionId.TryGetValue(subId, out var entry)
&& entry.FullRefByItemHandle.TryGetValue(handle, out var liveRef)
&& string.Equals(liveRef, fullRef, StringComparison.OrdinalIgnoreCase))
return handle;
return null;
}
/// <summary>Snapshot every active binding for diagnostic output.</summary>
/// <returns>A flattened list of every tag binding across all tracked subscriptions.</returns>
@@ -177,10 +136,6 @@ internal sealed class SubscriptionRegistry
if (remaining.IsEmpty)
{
_subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
// Clean up the forward index only when this handle is still the current
// mapping — the add loop below will overwrite it with the fresh handle.
if (_itemHandleByFullRef.TryGetValue(binding.FullReference, out var fwd) && fwd == binding.ItemHandle)
_itemHandleByFullRef.TryRemove(binding.FullReference, out _);
}
else _subscribersByItemHandle[binding.ItemHandle] = remaining;
}
@@ -193,7 +148,6 @@ internal sealed class SubscriptionRegistry
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
_itemHandleByFullRef[binding.FullReference] = binding.ItemHandle;
}
}