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

fix(galaxy): seam hardening - advise unmark on throw, dup attributeRef surfaced, bounded subscribe-once read; delete dormant historian connection probe (#605)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 06:52:21 -04:00
parent 3baa332d6b
commit 7c4c7f9653
8 changed files with 132 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+21 -2
View File
@@ -270,7 +270,17 @@ public sealed class GalaxyDriver
continue;
}
forward[entry.RawPath] = attributeRef;
reverse.TryAdd(attributeRef, entry); // first-wins on a shared attributeRef
if (!reverse.TryAdd(attributeRef, entry)) // first-wins on a shared attributeRef
{
// The loser is still dialled (forward map) but its value fan-out / alarm ConditionId surface
// under the winner's RawPath, so its own node never updates -- say so (archreview 06/S-13).
var winner = reverse[attributeRef].RawPath;
skipped.Skip(entry.RawPath,
$"duplicate attributeRef '{attributeRef}' — value fan-out surfaces under '{winner}'");
logger.LogWarning(
"GalaxyDriver raw tag {RawPath} shares attributeRef {AttributeRef} with {Winner} — its values and alarms surface under the winner only.",
entry.RawPath, attributeRef, winner);
}
}
skipped.Commit();
return (forward, reverse);
@@ -816,9 +826,18 @@ public sealed class GalaxyDriver
/// caller's cancellation token fires, return a Bad-status snapshot in input order
/// so the caller still sees one snapshot per requested reference.
/// </summary>
/// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
private async Task<IReadOnlyList<DataValueSnapshot>> ReadViaSubscribeOnceAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
IReadOnlyList<string> fullReferences, CancellationToken callerToken)
{
// Internal deadline (archreview 06/S-3): a tag that never publishes must not hang a caller that
// passed CancellationToken.None. Still-pending slots fill with BadTimeout exactly as for a caller cancel.
using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(callerToken);
deadlineCts.CancelAfter(ReadViaSubscribeDeadline);
var cancellationToken = deadlineCts.Token;
var pump = EnsureEventPumpStarted();
var subscriptionId = _subscriptions.NextSubscriptionId();
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs
+21 -1
View File
@@ -248,7 +248,27 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
},
};
var reply = await ops.InvokeAsync(request, ct).ConfigureAwait(false);
MxCommandReply reply;
try
{
reply = await ops.InvokeAsync(request, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
// The mark was taken before the call; a cancelled/thrown advise never committed, so forget it
// or every later write would skip the advise for the life of the session (archreview 06/S-12).
_supervisedHandles.TryRemove(itemHandle, out _);
throw;
}
catch (Exception ex)
{
_supervisedHandles.TryRemove(itemHandle, out _);
_logger.LogWarning(ex,
"GalaxyDriver supervisory advise threw for item {ItemHandle} — handle forgotten, write fails closed.",
itemHandle);
return false;
}
if (reply.ProtocolStatus is { } proto && proto.Code != ProtocolStatusCode.Ok)
{
// Supervisory advise failed — forget it so the next write retries the advise, and report
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs
+6 -52
View File
@@ -23,13 +23,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
/// </remarks>
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
{
/// <summary>
/// <see cref="ConnectionStatus.ConnectionKind"/> is a combinable [Flags] value: the
/// process-data connection is bit 0 (value 1), the event connection is bit 1 (value 2).
/// </summary>
private const uint ProcessConnectionFlag = 1;
private const uint EventConnectionFlag = 2;
private readonly IHistorianGatewayClient _client;
private readonly ILogger<GatewayHistorianDataSource> _logger;
@@ -41,8 +34,6 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
private long _totalSuccesses;
private long _totalFailures;
private int _consecutiveFailures;
private bool _processConnectionOpen;
private bool _eventConnectionOpen;
/// <summary>Creates a gateway-backed historian data source.</summary>
/// <param name="client">The gateway client seam used for all reads.</param>
@@ -194,55 +185,18 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
LastSuccessTime: _lastSuccessUtc,
LastFailureTime: _lastFailureUtc,
LastError: _lastError,
// Cached connection flags last observed by RefreshConnectionStateAsync. The gateway
// is non-clustered to us, so node fields are null/empty (mirrors the Wonderware
// client's Finding 010 posture).
ProcessConnectionOpen: _processConnectionOpen,
EventConnectionOpen: _eventConnectionOpen,
// No connection probe exists (the old RefreshConnectionStateAsync had no production
// caller and was deleted, archreview 06/S-6): "open" is derived from observed reads
// alone -- at least one success and no failure since. The gateway is non-clustered to
// us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture).
ProcessConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
EventConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
ActiveProcessNode: null,
ActiveEventNode: null,
Nodes: []);
}
}
/// <summary>
/// Refreshes the cached process / event connection flags by querying the gateway's
/// connection status. Intended to be driven by a periodic health hosted-service, keeping
/// <see cref="GetHealthSnapshot"/> pure observation (it never performs I/O). The flags are
/// derived from <see cref="ConnectionStatus.ConnectedToServer"/> AND the matching
/// <see cref="ConnectionStatus.ConnectionKind"/> flag bit. A failed status query is a health
/// probe — it never throws to the caller; both flags degrade to closed until the next
/// successful refresh.
/// </summary>
/// <param name="cancellationToken">A token to cancel the status query.</param>
/// <returns>A task that completes when the cached flags have been updated.</returns>
public async Task RefreshConnectionStateAsync(CancellationToken cancellationToken)
{
bool processOpen;
bool eventOpen;
try
{
var status = await _client.GetConnectionStatusAsync(cancellationToken).ConfigureAwait(false);
var connected = status.ConnectedToServer;
processOpen = connected && (status.ConnectionKind & ProcessConnectionFlag) != 0;
eventOpen = connected && (status.ConnectionKind & EventConnectionFlag) != 0;
}
catch (Exception)
{
// A health probe must never crash the host; an unreachable gateway degrades both
// connection flags to closed until the next successful refresh.
_logger.LogDebug("Historian gateway connection-status refresh failed; treating both connections as closed.");
processOpen = false;
eventOpen = false;
}
lock (_healthLock)
{
_processConnectionOpen = processOpen;
_eventConnectionOpen = eventOpen;
}
}
/// <summary>
/// Reconciles a gateway at-time reply against the requested timestamps to honour the
/// <see cref="IHistorianDataSource.ReadAtTimeAsync"/> contract: exactly one snapshot per