feat(galaxy): wait on the gateway's stream-attached ack instead of the 1 s best-effort wait (#877)

mxaccessgw#227 (Contracts/Client 0.4.0) adds an opt-in attach acknowledgement on
StreamEvents: a StreamAttached sentinel MxEvent sent first, only after the
stream's subscriber is registered on the session. This closes the #865 attach
race that waiting for "stream requested" only narrowed.

- Bump vendored MxGateway.Client/Contracts 0.3.0 -> 0.4.0 (nuget-packages/,
  pulled from the Gitea feed; local-mxgw stays the mapped source).
- IGalaxySubscriber.SupportsStreamAttachedAck (default false, so fakes and older
  seams keep the fallback); GatewayGalaxySubscriber reads it from the live
  session's capabilities (stream-events-attached-ack) and passes
  emitStreamAttached accordingly; TracedGalaxySubscriber forwards it.
- EventPump: records AttachAckExpected before the stream opens (published
  before StreamRequested completes); intercepts ev.StreamAttached ahead of the
  ReplayGap check - completes StreamAttached and continues: never dispatched,
  never counted as received, never a resume cursor (worker_sequence 0). The
  signal also completes when the loop ends so no waiter hangs. One pump = one
  stream, so a reconnect's new pump has a fresh signal. New counter
  galaxy.events.stream_attached.
- AwaitCurrentEventStreamAsync: with the capability, waits on StreamAttached
  within the same StreamRequestWait budget (1 s); a missing ack logs a
  rate-limited Warning and proceeds instead of hanging. Without the capability
  the old StreamRequested wait is the fallback.

Tests (StreamAttachedAckTests, 6): ack-gated subscribe gets its initial value;
capability absent uses the fallback; missing ack times out with a warning; the
sentinel is never dispatched; a dead stream releases ack waiters; reconnect
gives each stream a fresh signal. With the ack wait skipped, 3 of them fail.

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs

@@ -2213,9 +2213,11 @@ public sealed class GalaxyDriver
     }
 
     /// <summary>
-    ///     Upper bound on how long an advise waits for a current-session event stream (#865). Kept well under
-    ///     Galaxy's Tier-A Read (2 s) / Subscribe (5 s) call budgets, so the "proceed anyway" fallback is
-    ///     reachable before the resilience pipeline times the call out.
+    ///     Upper bound on how long an advise waits for a current-session event stream (#865) — including, on a
+    ///     gateway with the stream-attached ack, the ack itself (#877). Kept well under Galaxy's Tier-A Read (2 s) /
+    ///     Subscribe (5 s) call budgets, so the "proceed anyway" fallback is reachable before the resilience
+    ///     pipeline times the call out. The ack normally lands within one round trip, so a timeout on that path is
+    ///     a fault worth a warning, not a routine fallback.
     /// </summary>
     internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1);
 
@@ -2227,9 +2229,14 @@ public sealed class GalaxyDriver
     ///     <para>
     ///     A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before
     ///     the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish
-    ///     again for a long time. So: start the pump if there is none, and wait until its stream call has started
-    ///     (<see cref="EventPump.StreamRequested"/> — this narrows, it cannot close, the attach race: the gateway
-    ///     offers no attach acknowledgement).
+    ///     again for a long time. So: start the pump if there is none, and wait until its stream is attached.
+    ///     </para>
+    ///     <para>
+    ///     #877 — when the gateway advertised <c>stream-events-attached-ack</c> (<see cref="EventPump.AttachAckExpected"/>),
+    ///     "attached" is deterministic: the wait is on <see cref="EventPump.StreamAttached"/>, the gateway's sentinel
+    ///     that the stream is registered on the session, so the advise cannot lose its initial value. An older
+    ///     gateway has no ack; the wait falls back to <see cref="EventPump.StreamRequested"/> (the stream call has
+    ///     started), which only narrows the attach race.
     ///     </para>
     ///     <para>
     ///     A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen →
@@ -2284,6 +2291,22 @@ public sealed class GalaxyDriver
                     try
                     {
                         await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
+                        // AttachAckExpected is published before StreamRequested completes (#877).
+                        if (pump.AttachAckExpected && !pump.HasEnded)
+                        {
+                            remaining = StreamRequestWait - elapsed.Elapsed;
+                            if (remaining < TimeSpan.Zero) remaining = TimeSpan.Zero;
+                            try
+                            {
+                                await pump.StreamAttached.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
+                            }
+                            catch (TimeoutException)
+                            {
+                                if (pump.HasEnded) ReportEndedPump(pump);
+                                LogNoStreamAttachedAck();
+                                return session;
+                            }
+                        }
                         if (pump.HasEnded) ReportEndedPump(pump);
                         return session;
                     }
@@ -2292,8 +2315,10 @@ public sealed class GalaxyDriver
                         // fall through to the log
                     }
                 }
-                else if (pump.StreamRequested.IsCompleted)
+                else if (pump.StreamRequested.IsCompleted
+                         && (!pump.AttachAckExpected || pump.StreamAttached.IsCompleted))
                 {
+                    if (pump.HasEnded) ReportEndedPump(pump);
                     return session;
                 }
             }
@@ -2340,6 +2365,30 @@ public sealed class GalaxyDriver
     }
 
     private long _lastNoStreamWarningTicks;
+    private long _lastNoAckWarningTicks;
+
+    /// <summary>
+    ///     #877 — the gateway advertised the stream-attached ack but did not send it within
+    ///     <see cref="StreamRequestWait"/>. Loud (Warning, once a minute) because the capability promises the
+    ///     sentinel as the stream's first message: its absence means a slow/stuck attach or a gateway bug, and the
+    ///     advise proceeding now may lose its initial value.
+    /// </summary>
+    private void LogNoStreamAttachedAck()
+    {
+        var now = Environment.TickCount64;
+        var last = Interlocked.Read(ref _lastNoAckWarningTicks);
+        if ((last == 0 || now - last >= 60_000)
+            && Interlocked.CompareExchange(ref _lastNoAckWarningTicks, now, last) == last)
+        {
+            _logger.LogWarning(
+                "GalaxyDriver {InstanceId}: the gateway advertises stream-events-attached-ack but its event stream sent no attach acknowledgement within {Wait} — advising anyway; initial values may be missed.",
+                _driverInstanceId, StreamRequestWait);
+            return;
+        }
+        _logger.LogDebug(
+            "GalaxyDriver {InstanceId} advising without the event stream's attach acknowledgement (warning rate-limited).",
+            _driverInstanceId);
+    }
 
     /// <summary>Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood).</summary>
     private void LogNoCurrentStream()

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs

@@ -51,6 +51,9 @@ internal sealed class EventPump : IAsyncDisposable
     private static readonly Counter<long> ReplayGapsObserved =
         Meter.CreateCounter<long>("galaxy.events.replay_gaps", unit: "{gap}",
             description: "Replay-gap sentinels received — the gateway evicted events we can never see.");
+    private static readonly Counter<long> StreamAttachedAcks =
+        Meter.CreateCounter<long>("galaxy.events.stream_attached", unit: "{ack}",
+            description: "Stream-attached sentinels received — the gateway confirmed the event stream is registered (#877).");
     private static readonly Counter<long> ValuesUndecodable =
         Meter.CreateCounter<long>("galaxy.events.undecodable", unit: "{event}",
             description: "Data-change events carrying a value kind this driver cannot represent.");
@@ -175,10 +178,38 @@ internal sealed class EventPump : IAsyncDisposable
     ///     ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent
     ///     asynchronously and the gateway attaches the subscriber after auth, concurrently with any other
     ///     RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an
-    ///     MXAccess advise), but the gateway offers no attach acknowledgement to await.
+    ///     MXAccess advise). A gateway with the <c>stream-events-attached-ack</c> capability (mxaccessgw#227)
+    ///     closes it: wait on <see cref="StreamAttached"/> when <see cref="AttachAckExpected"/> (#877).
     /// </remarks>
     public Task StreamRequested => _streamRequested.Task;
 
+    private readonly TaskCompletionSource _streamAttached = new(TaskCreationOptions.RunContinuationsAsynchronously);
+    private volatile bool _attachAckExpected;
+
+    /// <summary>
+    ///     #877 — true when this pump's stream was requested WITH the gateway's stream-attached acknowledgement
+    ///     (the subscriber reported <see cref="IGalaxySubscriber.SupportsStreamAttachedAck"/> when the stream was
+    ///     opened). Set before <see cref="StreamRequested"/> completes, so a caller that awaited that first reads
+    ///     the final value.
+    /// </summary>
+    public bool AttachAckExpected => _attachAckExpected;
+
+    /// <summary>
+    ///     #877 — completes when the gateway's stream-attached sentinel arrives: the stream is registered on the
+    ///     gateway session, so every event the session publishes from then on (an advise's initial value
+    ///     included) reaches this stream. This CLOSES the attach race that <see cref="StreamRequested"/> only
+    ///     narrows. Also completes once the loop has ended, so a waiter never hangs on a dead pump — check
+    ///     <see cref="HasEnded"/> afterwards. Never completes for a stream opened without the ack
+    ///     (<see cref="AttachAckExpected"/> false) while it is alive: callers wait on it only when expected.
+    ///     One pump owns exactly one stream; a reconnect replaces the pump, so every new stream starts with a
+    ///     fresh, uncompleted signal.
+    /// </summary>
+    public Task StreamAttached => _streamAttached.Task;
+
+    /// <summary>True once the stream-attached sentinel itself arrived (not merely the loop ending).</summary>
+    internal bool StreamAttachedReceived => Volatile.Read(ref _attachAckReceived);
+    private bool _attachAckReceived;
+
     /// <summary>
     ///     Marker written into the channel to wake the dispatch loop after a registration promoted
     ///     parked early events (#859). Never dispatched itself; compared by reference.
@@ -213,6 +244,9 @@ internal sealed class EventPump : IAsyncDisposable
         IAsyncEnumerator<MxEvent>? enumerator = null;
         try
         {
+            // #877: read the capability BEFORE opening the stream — the subscriber requests the ack from the
+            // same session state — and publish it before StreamRequested releases a waiter.
+            _attachAckExpected = _subscriber.SupportsStreamAttachedAck;
             enumerator = _subscriber.StreamEventsAsync(ct).GetAsyncEnumerator(ct);
             // Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is
             // the moment StreamRequested reports (#865).
@@ -234,6 +268,17 @@ internal sealed class EventPump : IAsyncDisposable
             {
                 var ev = enumerator.Current;
                 if (ct.IsCancellationRequested) break;
+
+                // #877: the stream-attached sentinel is a control signal (family UNSPECIFIED, no body,
+                // worker_sequence 0). It is handled before anything else: it is never dispatched, never
+                // counted as a received/unknown event, and never touches a resume cursor (this driver
+                // keeps none — it always opens fresh streams — and a 0 sequence must never become one).
+                if (ev.StreamAttached is not null)
+                {
+                    HandleStreamAttached(ev.StreamAttached);
+                    continue;
+                }
+
                 EventsReceived.Add(1, _clientTag);
 
                 // The replay-gap sentinel is a control signal, not data, and is handled HERE
@@ -294,6 +339,8 @@ internal sealed class EventPump : IAsyncDisposable
             Volatile.Write(ref _ended, true);
             // A loop that died before issuing the RPC must not strand a StreamRequested waiter.
             _streamRequested.TrySetResult();
+            // Nor one waiting for an attach acknowledgement that can no longer arrive (#877).
+            _streamAttached.TrySetResult();
             if (enumerator is not null)
             {
                 try { await enumerator.DisposeAsync().ConfigureAwait(false); }
@@ -358,6 +405,25 @@ internal sealed class EventPump : IAsyncDisposable
     ///     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.
     /// </remarks>
+    private void HandleStreamAttached(ZB.MOM.WW.MxGateway.Contracts.Proto.StreamAttached attached)
+    {
+        StreamAttachedAcks.Add(1, _clientTag);
+        if (!_attachAckExpected)
+        {
+            // An unrequested ack (capability read raced a session swap, or a gateway that sends it anyway)
+            // is harmless — still a control signal, still never dispatched.
+            _logger.LogDebug("Galaxy event stream sent a stream-attached acknowledgement that was not expected — ignoring it.");
+        }
+        else
+        {
+            _logger.LogDebug(
+                "Galaxy event stream attached on the gateway (live after worker_sequence {LiveAfter}).",
+                attached.LiveAfterSequence);
+        }
+        Volatile.Write(ref _attachAckReceived, true);
+        _streamAttached.TrySetResult();
+    }
+
     private void HandleReplayGap(ReplayGap gap)
     {
         ReplayGapsObserved.Add(1, _clientTag);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxySubscriber.cs

@@ -158,6 +158,15 @@ public sealed class GatewayGalaxySubscriber : IGalaxySubscriber
         var session = _session.Session
             ?? throw new InvalidOperationException(
                 "GalaxyMxSession is not connected. StreamEventsAsync called before ConnectAsync.");
-        return session.StreamEventsAsync(afterWorkerSequence: 0, cancellationToken);
+        // #877: ask for the stream-attached sentinel only when the gateway advertised it — an older
+        // gateway ignores the field, so the pump would wait for an ack that never comes.
+        return session.StreamEventsAsync(
+            afterWorkerSequence: 0,
+            emitStreamAttached: session.SupportsStreamAttachedAck,
+            cancellationToken);
     }
+
+    /// <inheritdoc />
+    /// <remarks>Read from the live session's <c>OpenSessionReply.capabilities</c>; false while disconnected.</remarks>
+    public bool SupportsStreamAttachedAck => _session.Session?.SupportsStreamAttachedAck ?? false;
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/IGalaxySubscriber.cs

@@ -38,4 +38,15 @@ public interface IGalaxySubscriber
     /// <param name="cancellationToken">Cancellation token for the stream.</param>
     /// <returns>An asynchronous stream of <see cref="MxEvent"/>s as they arrive from the gateway.</returns>
     IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken);
+
+    /// <summary>
+    ///     #877 — true when the stream returned by <see cref="StreamEventsAsync"/> opens with the gateway's
+    ///     stream-attached sentinel (<see cref="MxEvent.StreamAttached"/>, mxaccessgw#227): the session
+    ///     advertised the <c>stream-events-attached-ack</c> capability and the subscriber requests the ack.
+    ///     The sentinel means the stream is registered on the gateway session, so an advise issued after it
+    ///     cannot lose its initial value to the attach race (#865). False (the default) for an older gateway
+    ///     and for test fakes that do not model the ack — the <see cref="EventPump"/> then falls back to the
+    ///     best-effort "stream requested" wait.
+    /// </summary>
+    bool SupportsStreamAttachedAck => false;
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/TracedGalaxySubscriber.cs

@@ -49,6 +49,9 @@ internal sealed class TracedGalaxySubscriber(IGalaxySubscriber inner, string cli
         }
     }
 
+    /// <inheritdoc />
+    public bool SupportsStreamAttachedAck => inner.SupportsStreamAttachedAck;
+
     // Streaming RPC — one parent span covers the entire stream lifetime. Per-event
     // spans would dominate the trace volume at 50k tags / 1Hz; ops gets per-event
     // visibility through EventPump's metrics instead.