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

@@ -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.

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.
     /// </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

@@ -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

@@ -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
+                // No connection probe exists (the old RefreshConnectionStateAsync had no production
-                // is non-clustered to us, so node fields are null/empty (mirrors the Wonderware
+                // caller and was deleted, archreview 06/S-6): "open" is derived from observed reads
-                // client's Finding 010 posture).
+                // alone -- at least one success and no failure since. The gateway is non-clustered to
-                ProcessConnectionOpen: _processConnectionOpen,
+                // us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture).
-                EventConnectionOpen: _eventConnectionOpen,
+                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

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs

@@ -257,4 +257,26 @@ public sealed class GalaxyDriverRawPathKeyingTests
             return Task.CompletedTask;
         }
     }
+
+    /// <summary>Two RawPaths sharing one attributeRef: first wins the reverse map, the loser is recorded in the skipped ledger (06/S-13).</summary>
+    [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);
+    }
 }

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);
     }
+
+    /// <summary>A never-publishing tag with CancellationToken.None must complete with BadTimeout at the internal deadline (06/S-3).</summary>
+    [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"]);
+    }
 }

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
     }
 
+    /// <summary>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.</summary>
+    [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);
+    }
+
     /// <summary>When advise succeeds, the raw Write proceeds and a clean reply returns Good (regression).</summary>
     [Fact]
     public async Task WriteOne_AdviseSucceeds_RawWriteProceeds()

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<GatewayHistorianDataSource>.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
+        var fake = new FakeHistorianGatewayClient { RawSamples = Array.Empty<HistorianSample>() };
-        {
-            ConnectionStatus = new ConnectionStatus { ConnectedToServer = false, ConnectionKind = 0b11 },
-        };
         var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
-        await ds.RefreshConnectionStateAsync(TestContext.Current.CancellationToken);
+        await ds.ReadRawAsync("T", default, default, 1, TestContext.Current.CancellationToken);
-        var h = ds.GetHealthSnapshot();
+        Assert.True(ds.GetHealthSnapshot().ProcessConnectionOpen);
-        Assert.False(h.ProcessConnectionOpen);
+        fake.ThrowOnRead = new InvalidOperationException("boom");
-        Assert.False(h.EventConnectionOpen);
+        await Assert.ThrowsAnyAsync<Exception>(() => ds.ReadRawAsync("T", default, default, 1, TestContext.Current.CancellationToken));
-    }
-
-    [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<GatewayHistorianDataSource>.Instance);
-        await ds.RefreshConnectionStateAsync(TestContext.Current.CancellationToken); // must not throw
         var h = ds.GetHealthSnapshot();
         Assert.False(h.ProcessConnectionOpen);
         Assert.False(h.EventConnectionOpen);
+        Assert.Equal(0, fake.GetConnectionStatusCallCount); // never probes the gateway
     }
 }