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

fix(galaxy): #865 review round 2 - wait for the replay's pump, never race a second stream

Browse Source 
Round 1 replaced a stale-session pump from any advise path and disposed the
old one in the background; that could start a second stream while the old
one still held the gateway's single-subscriber slot, and a kept
current-session pump could be dead with no fault re-reported.

- EnsureEventPumpStarted reuses again; only ReplayAsync restarts the pump.
  RestartEventPumpForReplay keeps a current-session pump only while its loop
  is running (EventPump.HasEnded), and flags the restart so no advise path
  starts a competing pump in the dispose gap.
- AwaitCurrentEventStreamAsync (bounded by StreamRequestWait) starts a pump
  only when none exists, WAITS for the replay while the pump is stale or
  restarting, waits for StreamRequested, and returns the session it verified.
  SubscribeAsync (per #861 attempt), read-via-subscribe and the probe hook
  issue and fence the advise under that session, so a reconnect after the
  check is caught as a straddle and retried.
- Read: a deadline (not caller) cancel during the wait falls through to
  per-ref results.
- Tests: reopen-window probe test now asserts the advise waits for the
  replay's single new stream; first Subscribe/Read wait for the stream; the
  fencing tests that advise in the window use a short wait bound; the
  production-path seed test uses an empty galaxy (a real probe advise now
  starts a stream against the fake endpoint).

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 12:46:15 -05:00
parent cc1a7ec687
commit 9060ff5561
6 changed files with 199 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+129 -81
View File
@@ -1056,7 +1056,7 @@ public sealed class GalaxyDriver
// #865: every re-advise below pushes its initial value once, to whatever stream is attached.
EventPump? replayPump;
lock (_pumpLock) replayPump = _eventPump;
if (replayPump is not null) await AwaitStreamRequestedAsync(replayPump, cancellationToken).ConfigureAwait(false);
if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
// Only subscriptions still holding the previous session's handles: one registered after the
// reopen is already live on the recreated session (re-subscribing it would double-advise).
@@ -1312,31 +1312,45 @@ public sealed class GalaxyDriver
{
old = _eventPump;
if (old is null) return; // pump never started — nothing to restart
// #865: a pump created after the reopen's fence (an advise in the reopen → replay window
// started it via EnsureEventPumpStarted) already streams the recreated session, and advises
// registered under it may be waiting for their initial values on it — keep it.
if (old.Session == _subscriptions.CurrentSession) return;
// #865: a pump first started after the reopen's fence (no pump existed before the reconnect)
// already streams the recreated session, and advises registered under it may be waiting for
// their initial values on it — keep it, unless its loop has ended (a dead pump is always replaced,
// #859 P-1: a stream fault during Replaying is not re-reported by the supervisor).
if (old.Session == _subscriptions.CurrentSession && !old.HasEnded) return;
_eventPump = null;
// #865: until the replacement exists, AwaitCurrentEventStreamAsync must not start one of its
// own — two streams would race for the gateway's single-subscriber slot while the old one detaches.
_pumpRestarting = true;
}
// Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't
// block a concurrent EnsureEventPumpStarted; the old loop already exited on fault.
old.OnDataChange -= OnPumpDataChange;
try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); }
catch (Exception ex)
try
{
_logger.LogWarning(ex,
"GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.",
_driverInstanceId);
}
// Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't
// block a concurrent EnsureEventPumpStarted; the old loop already exited on fault.
old.OnDataChange -= OnPumpDataChange;
try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); }
catch (Exception ex)
{
_logger.LogWarning(ex,
"GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.",
_driverInstanceId);
}
// EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the
// recreate if the driver is being disposed or its runtime torn down — the teardown
// already released the pump and a fresh one here would leak past it.
if (_disposed || _runtimeStopping) return;
EnsureEventPumpStarted();
// EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the
// recreate if the driver is being disposed or its runtime torn down — the teardown
// already released the pump and a fresh one here would leak past it.
if (_disposed || _runtimeStopping) return;
EnsureEventPumpStarted();
}
finally
{
lock (_pumpLock) _pumpRestarting = false;
}
}
/// <summary>True while <see cref="RestartEventPumpForReplay"/> is between dropping the old pump and starting its replacement.</summary>
private bool _pumpRestarting;
private void OnSupervisorStateChanged(object? sender, StateTransition transition)
{
// Reflect supervisor state in DriverHealth + transport forwarder.
@@ -1689,10 +1703,19 @@ public sealed class GalaxyDriver
var readSubscriber = _subscriber!;
// A live pump must exist (values arrive through it); the read listens on ReadDataChange, not
// on this instance, so it survives a reconnect's pump restart.
var readPump = EnsureEventPumpStarted();
// #865: a read waits ONLY for the initial value, which reaches only an already-requested stream.
// Not the caller's token: a cancel here must still yield per-ref Bad snapshots, and the wait is bounded.
await AwaitStreamRequestedAsync(readPump, CancellationToken.None).ConfigureAwait(false);
EnsureEventPumpStarted();
// #865: a read waits ONLY for the initial value, which reaches only a current-session stream that is
// already requested. The internal deadline bounds this too; a deadline (not caller) cancel falls
// through so the read still answers per-ref, as it does for a deadline during the RPC.
int issueSession;
try
{
issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
{
issueSession = _subscriptions.CurrentSession;
}
var subscriptionId = _subscriptions.NextSubscriptionId();
// Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange
@@ -1729,7 +1752,7 @@ public sealed class GalaxyDriver
// window before the RPC and close it right after registration (the finally below), not after the
// value wait: a window held for the whole wait would park every unrelated unknown-handle event.
var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified)
try
{
// The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
@@ -1938,8 +1961,6 @@ public sealed class GalaxyDriver
var generation = Volatile.Read(ref _runtimeGeneration);
var subscriber = _subscriber;
var pump = EnsureEventPumpStarted();
// #865: the advise's initial value reaches only an already-requested stream (no replay).
await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false);
var subscriptionId = _subscriptions.NextSubscriptionId();
if (fullReferences.Count == 0)
@@ -1972,6 +1993,10 @@ public sealed class GalaxyDriver
for (var attempt = 1; ; attempt++)
{
// #865: per attempt (a #861 retry is a new session): the advise's initial value reaches only a
// current-session stream that is already requested — the gateway does not replay a fresh stream.
var issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
// #859: MXAccess's initial value for a new handle can reach the event pump before this
// reply is registered (the gateway gives no reply-before-event ordering, and the reply and
// the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events
@@ -1980,7 +2005,7 @@ public sealed class GalaxyDriver
var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
try
{
var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified)
var results = await subscriber
.SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken)
.ConfigureAwait(false);
@@ -2094,18 +2119,8 @@ public sealed class GalaxyDriver
/// <see cref="SubscriptionRegistry"/> reverse map, not by spinning a pump per
/// subscription.
/// </summary>
/// <remarks>
/// #865: a pump created for an OLDER registry session is replaced, not reused. Between
/// <see cref="ReopenAsync"/>'s session swap and <see cref="ReplayAsync"/>'s pump restart the
/// current pump still streams the dead gateway session; an advise made in that window goes to the
/// recreated session, so its initial value would reach no stream — and, registered under the
/// current session, the replay would not re-advise it either. A pump whose session is current
/// was created after the reopen's second fence, so its stream is on the recreated session.
/// </remarks>
private EventPump EnsureEventPumpStarted()
{
EventPump? stale = null;
EventPump pump;
lock (_pumpLock)
{
// Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this
@@ -2117,12 +2132,7 @@ public sealed class GalaxyDriver
throw new InvalidOperationException(
"GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync.");
}
if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump;
if (_eventPump is not null)
{
stale = _eventPump;
stale.OnDataChange -= OnPumpDataChange;
}
if (_eventPump is not null) return _eventPump;
_eventPump = new EventPump(
_subscriber!, _subscriptions, _logger,
channelCapacity: _options.MxAccess.EventPumpChannelCapacity,
@@ -2131,27 +2141,13 @@ public sealed class GalaxyDriver
onReplayGap: OnEventPumpReplayGap);
_eventPump.OnDataChange += OnPumpDataChange;
_eventPump.Start();
pump = _eventPump;
return _eventPump;
}
if (stale is not null) DisposeStalePumpInBackground(stale);
return pump;
}
/// <summary>Dispose a replaced pump off the caller's path (its stream is dead; the dispose only joins its loops).</summary>
private void DisposeStalePumpInBackground(EventPump stale) => _ = Task.Run(async () =>
{
try { await stale.DisposeAsync().ConfigureAwait(false); }
catch (Exception ex)
{
_logger.LogWarning(ex, "GalaxyDriver {InstanceId} stale-session EventPump dispose failed.", _driverInstanceId);
}
});
/// <summary>
/// <see cref="EnsureEventPumpStarted"/> for infrastructure callers (the probe watcher) that
/// must not fail because the runtime is going away: null — and no pump — while a teardown is in
/// progress or there is no subscriber.
/// <see cref="EnsureEventPumpStarted"/> for infrastructure callers that must not fail because the
/// runtime is going away: null — and no pump — while a teardown is in progress or there is no subscriber.
/// </summary>
/// <returns>The running (or just started) pump; null when none may be started.</returns>
private EventPump? TryEnsureEventPumpStarted()
@@ -2172,43 +2168,95 @@ public sealed class GalaxyDriver
}
}
/// <summary>Upper bound on how long an advise waits for the pump's StreamEvents call to be started.</summary>
/// <summary>Upper bound on how long an advise waits for a current-session event stream (#865).</summary>
internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// #865 — wait, bounded, until <paramref name="pump"/> has started its StreamEvents call, so an
/// advise made next has a stream for its initial value. A fresh stream (after_worker_sequence = 0)
/// is never replayed, so an initial value pushed before it attaches is lost — and a slow-changing
/// attribute (ScanState, a setpoint) may not publish again for a long time. On timeout the caller
/// proceeds (logged): a late stream is no worse than the pre-#865 behaviour.
/// #865 — make sure an advise about to be issued has an event stream for its initial value, and return
/// the registry session to issue (and register) it under.
/// </summary>
private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken)
/// <remarks>
/// <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).
/// </para>
/// <para>
/// A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen →
/// replay window): an advise now would go to the recreated session with no stream for it, and —
/// registered under the current session — the replay would not re-advise it. The pump is not replaced
/// here (only <see cref="ReplayAsync"/> restarts it, so two streams never race for the single-subscriber
/// slot); the advise WAITS for the replay's pump instead. The returned session is the one the pump was
/// verified against: a reconnect after this check makes the caller's registration fail as a session
/// straddle (#861), whose retry comes back through here.
/// </para>
/// <para>
/// Bounded by <see cref="StreamRequestWait"/>; on timeout the caller proceeds with the current session
/// (logged) — no worse than the pre-#865 behaviour.
/// </para>
/// </remarks>
/// <param name="cancellationToken">Cancels the wait.</param>
/// <returns>The registry session to issue the advise under.</returns>
private async Task<int> AwaitCurrentEventStreamAsync(CancellationToken cancellationToken)
{
try
{
await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException)
var elapsed = System.Diagnostics.Stopwatch.StartNew();
while (true)
{
var session = _subscriptions.CurrentSession;
EventPump? pump;
bool restarting;
lock (_pumpLock)
{
restarting = _pumpRestarting;
pump = restarting ? null : TryEnsureEventPumpStarted();
}
if (pump is null && !restarting) return session; // tearing down — the caller's own fences decide
var remaining = StreamRequestWait - elapsed.Elapsed;
if (pump is not null && pump.Session == session)
{
if (remaining > TimeSpan.Zero)
{
try
{
await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
return session;
}
catch (TimeoutException)
{
// fall through to the warning
}
}
else if (pump.StreamRequested.IsCompleted)
{
return session;
}
}
else if (remaining > TimeSpan.Zero)
{
// Reopen → replay window: the replay restarts the pump for the current session.
await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false);
continue;
}
_logger.LogWarning(
"GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.",
"GalaxyDriver {InstanceId} had no started event stream for the current gateway session within {Wait} — advising anyway; initial values may be missed.",
_driverInstanceId, StreamRequestWait);
return _subscriptions.CurrentSession;
}
}
/// <summary>
/// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait for
/// its stream (<see cref="AwaitStreamRequestedAsync"/>). A probe's initial ScanState value is usually
/// its only event, and a driver with no authored tags never Reads/Subscribes, so without this its
/// pump — and every host status — never existed.
/// #865 — the probe watcher's pre-subscribe hook (<see cref="AwaitCurrentEventStreamAsync"/>). A probe's
/// initial ScanState value is usually its only event, and a driver with no authored tags never
/// Reads/Subscribes, so without this its pump — and every host status — never existed.
/// </summary>
private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken)
private Task<int> EnsureProbeEventStreamAsync(CancellationToken cancellationToken)
{
// A seed pass a teardown already cancelled must not start a pump after the teardown finished.
cancellationToken.ThrowIfCancellationRequested();
var pump = TryEnsureEventPumpStarted();
if (pump is null) return;
await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false);
return AwaitCurrentEventStreamAsync(cancellationToken);
}
/// <summary>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs
+11 -6
View File
@@ -47,7 +47,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
// #865: invoked before every probe SubscribeBulk so the driver's event stream is attached when the
// gateway pushes the initial ScanState value (a fresh StreamEvents does not replay earlier events,
// and ScanState rarely changes — a lost initial value leaves the host missing indefinitely).
private readonly Func<CancellationToken, Task>? _ensureEventStream;
private readonly Func<CancellationToken, Task<int>>? _ensureEventStream;
// Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
// we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
@@ -86,8 +86,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable
/// <param name="bufferedUpdateIntervalMs">Buffered update interval in milliseconds; must be >= 0.</param>
/// <param name="ensureEventStream">
/// Optional: awaited before every probe SubscribeBulk to make sure the event stream that carries
/// the initial ScanState value is running and requested (the driver starts its
/// <see cref="EventPump"/> and waits for its StreamEvents call to be issued).
/// the initial ScanState value is running and requested for the current gateway session (the
/// driver starts its <see cref="EventPump"/> and waits for its StreamEvents call). Returns the
/// registry session the stream was verified against; the advise is issued and fenced under it.
/// </param>
internal PerPlatformProbeWatcher(
IGalaxySubscriber subscriber,
@@ -95,7 +96,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
SubscriptionRegistry? registry,
ILogger? logger = null,
int bufferedUpdateIntervalMs = 0,
Func<CancellationToken, Task>? ensureEventStream = null)
Func<CancellationToken, Task<int>>? ensureEventStream = null)
{
_subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
_aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator));
@@ -210,12 +211,16 @@ public sealed class PerPlatformProbeWatcher : IDisposable
// #865: the stream must be live BEFORE the advise — the initial value is the probe's only
// event until the platform's scan state changes, and an unattached stream never sees it.
if (_ensureEventStream is not null) await _ensureEventStream(cancellationToken).ConfigureAwait(false);
// The returned session is the one the stream was verified for: a reconnect after the check makes
// the session test below refuse the reply as a straddle, and the retry comes back through here.
int? streamSession = _ensureEventStream is null
? null
: await _ensureEventStream(cancellationToken).ConfigureAwait(false);
// #859: the initial ScanState value can overtake the SubscribeBulk reply; park it until
// PublishBindingsLocked registers the handles below (window closes at method exit).
using var earlyEventWindow = _registry?.BeginPendingSubscribe();
var session = CurrentSession; // the session this SubscribeBulk is issued on
var session = streamSession ?? CurrentSession; // the session this SubscribeBulk is issued on
// PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
// when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
// changes are rare so a coarser interval is usually fine; deployments that need
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
+3
View File
@@ -154,6 +154,9 @@ internal sealed class EventPump : IAsyncDisposable
/// </summary>
internal int Session => _session;
/// <summary>True once the stream loop has exited (faulted, completed by the gateway, or disposed).</summary>
internal bool HasEnded => _loop is { IsCompleted: true };
private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs
+4
View File
@@ -844,6 +844,7 @@ public sealed class EarlyEventWindowTests
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
{
RecreateOverrideForTests = _ => Task.CompletedTask,
StreamRequestWait = TimeSpan.FromMilliseconds(100), // advises in the reopen window on purpose (#865 bound)
};
var delivered = new ConcurrentQueue<(string Ref, double Value)>();
driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!));
@@ -1027,6 +1028,9 @@ public sealed class EarlyEventWindowTests
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
{
RecreateOverrideForTests = _ => Task.CompletedTask,
// These tests advise in the reopen → replay window on purpose (fencing, not values). #865 makes
// such an advise wait for the replay's pump, bounded; keep the bound short so they don't idle.
StreamRequestWait = TimeSpan.FromMilliseconds(100),
};
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
return driver;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs
+4 -1
View File
@@ -272,7 +272,10 @@ public sealed class GalaxyDriverFixRound1Tests
[Fact]
public async Task Production_path_queries_the_platform_source_at_init()
{
var platforms = new ScriptedPlatformSource();
// An empty galaxy: since #865 a probe advise starts the event stream first, and this owned runtime's
// gRPC channel points at no real gateway — its stream would fault and (correctly) degrade health.
// The query itself is what this test pins.
var platforms = new ScriptedPlatformSource { Script = _ => [] };
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: platforms)
{
SessionOpenOverrideForTests = _ => Task.CompletedTask,
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs
+48 -16
View File
@@ -195,10 +195,10 @@ public sealed class ProbeInitialValueTests
/// <summary>
/// Review finding: a probe advised in the reopen → replay window (the session was swapped, the old
/// pump still streams the dead session) must still get its initial value — the stale-session pump
/// is replaced before the advise, and the replay keeps the replacement (it already streams the
/// recreated session). Reusing the old pump bound the value to the old session, so it never routed,
/// and the replay skipped the platform (its handle is current-session).
/// pump still streams the dead session) must still get its initial value. The advise waits for the
/// replay's restarted pump instead of advising onto the stale one (its value would route under the
/// old session and be lost, and the replay skips a current-session handle) or starting a competing
/// stream (the gateway's default single-subscriber mode refuses a second attach).
/// </summary>
[Fact]
public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value()
@@ -218,23 +218,55 @@ public sealed class ProbeInitialValueTests
await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet
sub.ScanStateRunning = false;
await driver.ProbeWatcherForTests.ShouldNotBeNull()
var callsBeforeWindow = sub.CallsSnapshot().Length;
// The window advise must WAIT for the replay's current-session pump rather than advise onto the
// stale one (or start a competing stream).
var sync = driver.ProbeWatcherForTests.ShouldNotBeNull()
.SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None);
await TestWait.UntilAsync(
() => StateOf(driver, "LatePlatform") == HostState.Stopped,
"a probe advised in the reopen window must get its initial value",
() => Describe(driver, sub));
var windowPump = driver.EventPumpInstanceForTests;
windowPump.ShouldNotBeSameAs(oldPump, "the stale-session pump must have been replaced before the advise");
await driver.InvokeReplayForTestAsync(CancellationToken.None);
driver.EventPumpInstanceForTests.ShouldBeSameAs(windowPump, "the replay keeps a pump already on the current session");
await TestWait.ForAsync(sync, "the window advise completes once the replay restarted the pump");
await TestWait.UntilAsync(
() => Platforms.All(p => StateOf(driver, p) == HostState.Stopped),
"the replay re-advises the stale probes and their initial values arrive on the kept pump",
() => StateOf(driver, "LatePlatform") == HostState.Stopped
&& Platforms.All(p => StateOf(driver, p) == HostState.Stopped),
"the window advise and the replay's re-advises all get their initial values",
() => Describe(driver, sub));
StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped);
driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the replay restarted the stale pump");
var windowCalls = sub.CallsSnapshot()[callsBeforeWindow..];
var lateAdvise = Array.FindIndex(windowCalls, c => c.Contains("LatePlatform", StringComparison.Ordinal));
var newStream = Array.IndexOf(windowCalls, "stream");
newStream.ShouldBeGreaterThanOrEqualTo(0, $"the replay must start a new stream; calls=[{string.Join(" | ", windowCalls)}]");
lateAdvise.ShouldBeGreaterThan(newStream, $"the window advise must follow the new stream; calls=[{string.Join(" | ", windowCalls)}]");
windowCalls.Count(c => c == "stream").ShouldBe(1, "no competing stream may be started for the window advise");
sub.DroppedInitialValues.ShouldBe(0);
}
/// <summary>
/// The data paths too: a first SubscribeAsync and a first ReadAsync on a driver whose pump was never
/// started request the stream BEFORE advising, so their initial values arrive (a read of a value that
/// never changes again would otherwise time out).
/// </summary>
[Fact]
public async Task First_subscribe_and_read_advise_only_after_the_stream_is_requested()
{
var sub = new GatewayLikeSubscriber();
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
{
ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10),
};
await driver.InitializeAsync("{}", CancellationToken.None);
var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
driver.OnDataChange += (_, e) => delivered.TrySetResult(e);
await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
var calls = sub.CallsSnapshot();
Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the advise; calls=[{string.Join(" | ", calls)}]");
(await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub)))
.FullReference.ShouldBe("Tank01.Level");
var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None);
read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)");
sub.DroppedInitialValues.ShouldBe(0);
}