From 7c4c7f9653899b5c18acfae2f6acb415d97e0ac4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 06:52:21 -0400 Subject: [PATCH] fix(galaxy): seam hardening - advise unmark on throw, dup attributeRef surfaced, bounded subscribe-once read; delete dormant historian connection probe (#605) Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- docs/drivers/Galaxy.md | 3 + .../GalaxyDriver.cs | 23 +++++++- .../Runtime/GatewayGalaxyDataWriter.cs | 22 ++++++- .../GatewayHistorianDataSource.cs | 58 ++----------------- .../Runtime/GalaxyDriverRawPathKeyingTests.cs | 22 +++++++ .../Runtime/GalaxyDriverReadTests.cs | 21 +++++++ .../GatewayGalaxyDataWriterFailClosedTests.cs | 31 ++++++++++ .../GatewayHealthSnapshotTests.cs | 42 +++----------- 8 files changed, 132 insertions(+), 90 deletions(-) diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index da6525b1..e6a723c6 100644 --- a/docs/drivers/Galaxy.md +++ b/docs/drivers/Galaxy.md @@ -113,3 +113,6 @@ The session-less alarm feed (`GatewayGalaxyAlarmFeed`) and alarm acknowledger (` - **Replay gaps** (`galaxy.events.replay_gaps`): the gateway signals that events were evicted from its replay ring before we could read them. The driver responds by re-issuing `SubscribeBulk` for every tracked subscription — MXAccess delivers an initial value per item on subscribe, so re-subscribing *is* the re-snapshot the contract asks for. It does **not** restart the event pump (the stream carrying the sentinel is healthy) and it honours `Reconnect:ReplayOnSessionLost=false` by logging and standing down, on the grounds that an operator who delegated recovery to the gateway did not ask for a re-subscribe on a different trigger. ⚠️ This cannot currently fire: the sentinel is only emitted on a stream opened with `after_worker_sequence > 0`, and the driver opens with `0`, which the gateway treats as "fresh stream, not a resume". Enabling resume is a bigger change than it looks — the sequence is only meaningful **within one gateway session**, and the reconnect path recreates the session, so a remembered sequence replayed against a fresh session would set the server's live watermark *above* every event it is about to send, silencing the stream rather than recovering it. - **Undecodable values** (`galaxy.events.undecodable`): a value kind this driver has no CLR representation for is published as **`BadTypeMismatch` with a null value**, and the first occurrence of each distinct kind is logged. It is *not* published as a null carrying the event's own quality: an MXAccess event can be Good-quality while the payload is one we cannot map, and a good null is indistinguishable from a genuine null reading — a client would consume a value it never received as a real one, with the failure visible only in a log. Non-zero here means the gateway is sending a wire type `MxValueDecoder` has no mapping for and those readings are being lost; the fix is a decoder mapping, not a quality change. +- **Duplicate `attributeRef` (#605 / 06-S-13)**: two authored RawPaths dialling the same Galaxy attribute are both dialled, but the reverse map is first-wins, so value fan-out and alarm `ConditionId` surface under the first RawPath only. The loser is logged at Warning and recorded in the skipped-tag ledger with the winner path in the reason. +- **Subscribe-once reads are bounded (#605 / 06-S-3)**: `ReadAsync` on the production path (SubscribeBulk, first event, Unsubscribe) carries an internal 30 s deadline in addition to the caller token; slots that never publish come back `BadTimeout`, so `CancellationToken.None` can no longer hang on a silent tag. +- **Supervisory advise is retried after a thrown failure (#605 / 06-S-12)**: the writer marks an item handle advised before the call; a thrown advise (not just a non-OK status) now unmarks it, fails the write closed and re-advises on the next write. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 48fda20f..70615c41 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -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. /// + /// Upper bound on one subscribe-once read; internal so tests can shrink it. + internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30); + private async Task> ReadViaSubscribeOnceAsync( - IReadOnlyList fullReferences, CancellationToken cancellationToken) + IReadOnlyList 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(); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs index 9a25a321..7a5a894a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs @@ -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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 60db2223..033f5403 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -23,13 +23,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable { - /// - /// is a combinable [Flags] value: the - /// process-data connection is bit 0 (value 1), the event connection is bit 1 (value 2). - /// - private const uint ProcessConnectionFlag = 1; - private const uint EventConnectionFlag = 2; - private readonly IHistorianGatewayClient _client; private readonly ILogger _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; /// Creates a gateway-backed historian data source. /// The gateway client seam used for all reads. @@ -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: []); } } - /// - /// 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 - /// pure observation (it never performs I/O). The flags are - /// derived from AND the matching - /// 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. - /// - /// A token to cancel the status query. - /// A task that completes when the cached flags have been updated. - 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; - } - } - /// /// Reconciles a gateway at-time reply against the requested timestamps to honour the /// contract: exactly one snapshot per diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs index 69569dd9..4fec9f80 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs @@ -257,4 +257,26 @@ public sealed class GalaxyDriverRawPathKeyingTests return Task.CompletedTask; } } + + /// Two RawPaths sharing one attributeRef: first wins the reverse map, the loser is recorded in the skipped ledger (06/S-13). + [Fact] + public void DuplicateAttributeRef_RecordsLosingRawPath_InSkippedLedger() + { + var opts = OptsWithRawTags() with + { + RawTags = + [ + new RawTagEntry(RawPath, $"{{\"attributeRef\":\"{AttributeRef}\"}}", false), + new RawTagEntry("area/line/tank1/level-copy", $"{{\"attributeRef\":\"{AttributeRef}\"}}", false), + ], + }; + using var driver = new GalaxyDriver("g", opts, hierarchySource: null); + + var report = driver.GetSkippedTags(); + report.ShouldNotBeNull(); + report.AuthoredCount.ShouldBe(2); + report.SkippedCount.ShouldBe(1); + report.Sample[0].RawPath.ShouldBe("area/line/tank1/level-copy"); + report.Sample[0].Reason.ShouldContain(RawPath); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs index d8974f7b..9bb065df 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs @@ -158,4 +158,25 @@ public sealed class GalaxyDriverReadTests result[0].StatusCode.ShouldBe(StatusCodeMap.Good); result[1].StatusCode.ShouldBe(StatusCodeMap.BadNotConnected); } + + /// A never-publishing tag with CancellationToken.None must complete with BadTimeout at the internal deadline (06/S-3). + [Fact] + public async Task ReadAsync_SubscribeOncePath_NeverPublishes_NoCallerToken_TimesOutWithBadTimeout() + { + var subscriber = new GalaxyDriverSubscribeTests.FakeSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) + { + ReadViaSubscribeDeadline = TimeSpan.FromMilliseconds(200), + }; + + var readTask = driver.ReadAsync(["Silent.Tag"], CancellationToken.None); + var finished = await Task.WhenAny(readTask, Task.Delay(TimeSpan.FromSeconds(10))); + + finished.ShouldBe(readTask); + var result = await readTask; + result.Count.ShouldBe(1); + result[0].StatusCode.ShouldBe(StatusCodeMap.BadTimeout); + subscriber.UnsubscribedHandles.ShouldContain(subscriber.Map["Silent.Tag"]); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs index 5b514fba..67a01d77 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs @@ -62,6 +62,37 @@ public sealed class GatewayGalaxyDataWriterFailClosedTests writer.CachedSupervisedHandleCount.ShouldBe(0); // advise failure forgot the handle → next write retries } + /// A THROWN advise (transport fault, not a non-OK status) must also forget the handle, or every later + /// write on it would skip the advise forever (06/S-12). The next write re-advises and succeeds. + [Fact] + public async Task WriteOne_AdviseThrows_FailsClosed_ForgetsHandle_AndNextWriteReAdvises() + { + var throwNext = true; + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => throwNext + ? throw new IOException("transport reset") + : new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }, + }; + var writer = NewWriter(); + + var first = await writer.WriteOneForTestAsync( + ops, new WriteRequest("Tag.T", true), SecurityClassification.FreeAccess, CancellationToken.None); + + first.StatusCode.ShouldBe(StatusCodeMap.BadCommunicationError); + ops.RawWriteHandles.ShouldBeEmpty(); + writer.CachedSupervisedHandleCount.ShouldBe(0); + + throwNext = false; + var second = await writer.WriteOneForTestAsync( + ops, new WriteRequest("Tag.T", true), SecurityClassification.FreeAccess, CancellationToken.None); + + second.StatusCode.ShouldBe(StatusCodeMap.Good); + ops.Invokes.Count(r => r.Command.Kind == MxCommandKind.AdviseSupervisory).ShouldBe(2); + ops.RawWriteHandles.ShouldHaveSingleItem(); + writer.CachedSupervisedHandleCount.ShouldBe(1); + } + /// When advise succeeds, the raw Write proceeds and a clean reply returns Good (regression). [Fact] public async Task WriteOne_AdviseSucceeds_RawWriteProceeds() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHealthSnapshotTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHealthSnapshotTests.cs index 2992650a..d61086c0 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHealthSnapshotTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHealthSnapshotTests.cs @@ -22,20 +22,6 @@ public sealed class GatewayHealthSnapshotTests Assert.Equal(h.TotalQueries, h.TotalSuccesses + h.TotalFailures); // invariant } - [Fact] - public async Task Connection_state_reflects_GetConnectionStatus_flags() - { - var fake = new FakeHistorianGatewayClient - { - ConnectionStatus = new ConnectionStatus { ConnectedToServer = true, ConnectionKind = 0b11 }, // Process|Event - }; - var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); - await ds.RefreshConnectionStateAsync(TestContext.Current.CancellationToken); // internal probe used by health hosted-service - var h = ds.GetHealthSnapshot(); - Assert.True(h.ProcessConnectionOpen); - Assert.True(h.EventConnectionOpen); - } - [Fact] public async Task GetHealthSnapshot_does_no_io_and_starts_with_connections_closed() { @@ -49,31 +35,17 @@ public sealed class GatewayHealthSnapshotTests } [Fact] - public async Task Disconnected_status_leaves_both_flags_closed() + public async Task Connection_flags_are_derived_from_observed_reads() { - var fake = new FakeHistorianGatewayClient - { - ConnectionStatus = new ConnectionStatus { ConnectedToServer = false, ConnectionKind = 0b11 }, - }; + var fake = new FakeHistorianGatewayClient { RawSamples = Array.Empty() }; var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); - await ds.RefreshConnectionStateAsync(TestContext.Current.CancellationToken); - var h = ds.GetHealthSnapshot(); - Assert.False(h.ProcessConnectionOpen); - Assert.False(h.EventConnectionOpen); - } - - [Fact] - public async Task Failed_status_query_degrades_flags_without_throwing() - { - var fake = new FakeHistorianGatewayClient - { - ConnectionStatus = new ConnectionStatus { ConnectedToServer = true, ConnectionKind = 0b11 }, - GetConnectionStatusThrows = new InvalidOperationException("gateway unreachable"), - }; - var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); - await ds.RefreshConnectionStateAsync(TestContext.Current.CancellationToken); // must not throw + await ds.ReadRawAsync("T", default, default, 1, TestContext.Current.CancellationToken); + Assert.True(ds.GetHealthSnapshot().ProcessConnectionOpen); + fake.ThrowOnRead = new InvalidOperationException("boom"); + await Assert.ThrowsAnyAsync(() => ds.ReadRawAsync("T", default, default, 1, TestContext.Current.CancellationToken)); var h = ds.GetHealthSnapshot(); Assert.False(h.ProcessConnectionOpen); Assert.False(h.EventConnectionOpen); + Assert.Equal(0, fake.GetConnectionStatusCallCount); // never probes the gateway } }