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

fix(galaxy): #865 review round 3 - budget-safe stream wait, dead pump, fast-path drop

Browse Source 
- StreamRequestWait 1 s (under Tier-A Read 2 s / Subscribe 5 s budgets) and
  a stale pump is only waited on while the supervisor is reopening or
  replaying (or a restart is underway); in a reconnect backoff the advise
  proceeds at once. The no-stream log is a rate-limited Warning while
  Healthy, Debug otherwise.
- A current-session pump whose loop has ended is not a stream: if the
  supervisor already reports Healthy (its fault was swallowed while it was
  recovering) the failure is re-reported to start a new cycle.
- Subscribe/Read no longer start the pump directly (only
  AwaitCurrentEventStreamAsync does, honouring the replay restart flag);
  the teardown refusal is an explicit ThrowIfRuntimeStopping.
- SubscriptionRegistry.ResolveOrPark fast path: the handle lookup and the
  open-window check were not atomic, so a subscribe that registered the
  handle and closed its window in between got its initial value dropped as
  stale (found by a stress run: the first event of a bulk advise lost). It
  re-looks-up before declaring stale. Deterministic test via a seam.
- Tests: first Read on a fresh driver, replay of a data subscription.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 13:00:19 -05:00
parent 9060ff5561
commit 1c1705fa77
4 changed files with 189 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+101 -19
View File
@@ -1703,7 +1703,7 @@ 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.
EnsureEventPumpStarted();
ThrowIfRuntimeStopping(); // the pump itself is started by AwaitCurrentEventStreamAsync below
// #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.
@@ -1960,7 +1960,7 @@ public sealed class GalaxyDriver
// generation, so the Register below can tell its SubscribeBulk straddled a teardown.
var generation = Volatile.Read(ref _runtimeGeneration);
var subscriber = _subscriber;
var pump = EnsureEventPumpStarted();
ThrowIfRuntimeStopping(); // the pump is started per attempt by AwaitCurrentEventStreamAsync (#865)
var subscriptionId = _subscriptions.NextSubscriptionId();
if (fullReferences.Count == 0)
@@ -2014,7 +2014,6 @@ public sealed class GalaxyDriver
switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session))
{
case RegisterOutcome.Registered:
_ = pump; // keep the pump alive for the subscription's lifetime
return new GalaxySubscriptionHandle(subscriptionId);
case RegisterOutcome.TornDown:
@@ -2113,11 +2112,25 @@ public sealed class GalaxyDriver
}
}
/// <summary>A Subscribe/Read is refused while a teardown is in progress (the pump would belong to a dying subscriber).</summary>
private void ThrowIfRuntimeStopping()
{
lock (_pumpLock)
{
if (_runtimeStopping)
{
throw new InvalidOperationException(
"GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync.");
}
}
}
/// <summary>
/// Lazily start the <see cref="EventPump"/> on the first subscribe. The pump is
/// shared across every subscription on this driver — fan-out happens through the
/// <see cref="SubscriptionRegistry"/> reverse map, not by spinning a pump per
/// subscription.
/// Lazily start the <see cref="EventPump"/> on the first advise. Since #865 the advise paths
/// (probe seed, Subscribe, Read) go through <see cref="AwaitCurrentEventStreamAsync"/>, and only the
/// replay replaces an existing pump. The pump is shared across every subscription on this driver —
/// fan-out happens through the <see cref="SubscriptionRegistry"/> reverse map, not by spinning a pump
/// per subscription.
/// </summary>
private EventPump EnsureEventPumpStarted()
{
@@ -2168,8 +2181,12 @@ public sealed class GalaxyDriver
}
}
/// <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>
/// 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.
/// </summary>
internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1);
/// <summary>
/// #865 — make sure an advise about to be issued has an event stream for its initial value, and return
@@ -2180,20 +2197,30 @@ public sealed class GalaxyDriver
/// 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).
/// (<see cref="EventPump.StreamRequested"/> — this narrows, it cannot close, the attach race: the gateway
/// offers no attach acknowledgement).
/// </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.
/// here (only <see cref="ReplayAsync"/> restarts it, so two streams never race for the gateway's
/// single-subscriber slot); while the supervisor is actually reopening/replaying the advise WAITS for the
/// replay's pump. During a reconnect BACKOFF (TransportLost) no replay is imminent, so it does not wait.
/// 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.
/// Bounded by <see cref="StreamRequestWait"/> per call (a #861 retry starts a new budget). On timeout the
/// caller proceeds with the current session — no worse than the pre-#865 behaviour, but note the cost:
/// such an advise registers as current, so the replay will not re-advise it, and its initial value is
/// lost until the attribute next changes.
/// </para>
/// <para>
/// A current-session pump whose loop has ENDED is not a stream (its <see cref="EventPump.StreamRequested"/>
/// completes on exit). If the supervisor already reports Healthy, its fault was swallowed while it was
/// recovering (it ignores faults outside Healthy) — the driver would be deaf while Healthy — so it is
/// reported again to start a new recovery cycle.
/// </para>
/// </remarks>
/// <param name="cancellationToken">Cancels the wait.</param>
@@ -2216,16 +2243,22 @@ public sealed class GalaxyDriver
var remaining = StreamRequestWait - elapsed.Elapsed;
if (pump is not null && pump.Session == session)
{
if (pump.HasEnded)
{
ReportEndedPump(pump);
return session;
}
if (remaining > TimeSpan.Zero)
{
try
{
await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
if (pump.HasEnded) ReportEndedPump(pump);
return session;
}
catch (TimeoutException)
{
// fall through to the warning
// fall through to the log
}
}
else if (pump.StreamRequested.IsCompleted)
@@ -2233,18 +2266,67 @@ public sealed class GalaxyDriver
return session;
}
}
else if (remaining > TimeSpan.Zero)
else if (remaining > TimeSpan.Zero && ReplayExpectedSoon())
{
// Reopen → replay window: the replay restarts the pump for the current session.
await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false);
continue;
}
LogNoCurrentStream();
return _subscriptions.CurrentSession;
}
}
/// <summary>
/// True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is
/// no supervisor (the injected-seam path, where the replay is driven directly), or a restart is underway.
/// </summary>
private bool ReplayExpectedSoon()
{
if (Volatile.Read(ref _pumpRestarting)) return true;
var supervisor = _supervisor;
if (supervisor is null) return true;
var state = supervisor.CurrentState;
return state is ReconnectSupervisor.State.Reopening or ReconnectSupervisor.State.Replaying;
}
/// <summary>A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync).</summary>
private void ReportEndedPump(EventPump pump)
{
var supervisor = _supervisor;
if (supervisor is null || _runtimeStopping || supervisor.CurrentState != ReconnectSupervisor.State.Healthy) return;
lock (_pumpLock)
{
if (!ReferenceEquals(_eventPump, pump)) return; // already replaced
}
_logger.LogWarning(
"GalaxyDriver {InstanceId} event stream has ended while the reconnect supervisor reports Healthy — re-reporting the transport failure.",
_driverInstanceId);
try { supervisor.ReportTransportFailure(new InvalidOperationException("The Galaxy event stream ended; no events can be received.")); }
catch (ObjectDisposedException) { /* driver disposing */ }
}
private long _lastNoStreamWarningTicks;
/// <summary>Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood).</summary>
private void LogNoCurrentStream()
{
var supervisor = _supervisor;
var now = Environment.TickCount64;
var last = Interlocked.Read(ref _lastNoStreamWarningTicks);
if ((supervisor is null || supervisor.CurrentState == ReconnectSupervisor.State.Healthy)
&& (last == 0 || now - last >= 60_000)
&& Interlocked.CompareExchange(ref _lastNoStreamWarningTicks, now, last) == last)
{
_logger.LogWarning(
"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;
return;
}
_logger.LogDebug(
"GalaxyDriver {InstanceId} advising without a started current-session event stream (supervisor not Healthy or warning rate-limited).",
_driverInstanceId);
}
/// <summary>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
+18 -1
View File
@@ -314,6 +314,12 @@ internal sealed class SubscriptionRegistry
/// </summary>
internal Action? AfterMapWriteForTests { get; set; }
/// <summary>
/// Test seam (#865): invoked on <see cref="ResolveOrPark"/>'s lock-free fast path after the handle
/// lookup missed, before the open-window check.
/// </summary>
internal Action? AfterFastPathMissForTests { get; set; }
/// <summary>
/// Test seam (M-1): parks <paramref name="ev"/> unconditionally, producing the parked list for a
/// resolved handle that the atomic registration otherwise makes impossible.
@@ -496,7 +502,18 @@ internal sealed class SubscriptionRegistry
{
subscribers = ResolveSubscribers(ev.ItemHandle);
if (subscribers.Count > 0) return EarlyEventOutcome.Resolved;
if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale;
AfterFastPathMissForTests?.Invoke();
if (Volatile.Read(ref _pendingSubscribes) == 0)
{
// #865 (found by its stress run): the lookup above and this read are not atomic. A
// subscribe can register the handle AND close its window in between — then "no window
// open" does not mean "stale", and dropping here lost the initial value for good (the
// first event of a bulk advise, while the dispatch thread was slow). The window closes
// only after its registration, so once no window is open a re-lookup sees every
// registration that window made; nothing is held, so no older event can be overtaken.
subscribers = ResolveSubscribers(ev.ItemHandle);
return subscribers.Count > 0 ? EarlyEventOutcome.Resolved : EarlyEventOutcome.Stale;
}
}
lock (_parkGate)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs
+25
View File
@@ -898,6 +898,31 @@ public sealed class EarlyEventWindowTests
driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty();
}
/// <summary>
/// #865 stress finding: the dispatch fast path looked the handle up, then checked for an open window —
/// not atomically. A subscribe that registered the handle AND closed its window in between made the
/// event "stale" and its initial value was dropped for good. It must resolve.
/// </summary>
[Fact]
public void A_registration_landing_between_the_fast_path_lookup_and_the_window_check_still_resolves()
{
var registry = new SubscriptionRegistry();
var session = registry.CurrentSession;
var window = registry.BeginPendingSubscribe();
registry.AfterFastPathMissForTests = () =>
{
registry.AfterFastPathMissForTests = null;
registry.TryRegister(registry.NextSubscriptionId(), [new TagBinding("Tank.A", 7)], session).ShouldBeTrue();
window.Dispose(); // the subscribe's window closes right after its registration
};
registry.ResolveOrPark(DataChange(7, 1), session, out var subscribers, out _).ShouldBe(EarlyEventOutcome.Resolved);
subscribers.Single().FullReference.ShouldBe("Tank.A");
// Positive control for the Stale branch: an unknown handle with no window open is still dropped.
registry.ResolveOrPark(DataChange(8, 1), session, out _, out _).ShouldBe(EarlyEventOutcome.Stale);
}
/// <summary>I-B: a purge that completes after a reconnect does not delete the new session's parked values.</summary>
[Fact]
public void A_purge_for_an_older_session_is_a_no_op()
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs
+45
View File
@@ -265,11 +265,56 @@ public sealed class ProbeInitialValueTests
(await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub)))
.FullReference.ShouldBe("Tank01.Level");
sub.DroppedInitialValues.ShouldBe(0);
}
/// <summary>A first ReadAsync on a fresh driver (no pump yet) requests the stream before its advise.</summary>
[Fact]
public async Task First_read_advises_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);
driver.HasEventPumpForTests.ShouldBeFalse("precondition: the read is the first advise");
var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None);
var calls = sub.CallsSnapshot();
Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the read's advise; calls=[{string.Join(" | ", calls)}]");
read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)");
sub.DroppedInitialValues.ShouldBe(0);
}
/// <summary>
/// The replay's re-advise of DATA subscriptions (which has no per-advise hook) waits for the restarted
/// pump's stream, so each re-advised tag's new initial value arrives. The value flips so only a
/// post-replay event can satisfy the wait.
/// </summary>
[Fact]
public async Task Reconnect_replay_delivers_re_advised_data_subscriptions_initial_values()
{
var sub = new GatewayLikeSubscriber();
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
await driver.InitializeAsync("{}", CancellationToken.None);
var last = new System.Collections.Concurrent.ConcurrentDictionary<string, object?>();
driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value;
await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is true,
"positive control: the subscribe's initial value arrives", () => Describe(driver, sub));
sub.ScanStateRunning = false; // the fake's next initial value
await driver.InvokeReplayForTestAsync(CancellationToken.None);
await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false,
"the replay's re-advise must deliver its new initial value",
() => Describe(driver, sub) + $" last=[{string.Join(",", last.Select(kv => $"{kv.Key}={kv.Value}"))}] parked={driver.ParkedEarlyEventCountForTests} h2=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(2).Select(b => b.FullReference))}] h1=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(1).Select(b => b.FullReference))}]");
sub.DroppedInitialValues.ShouldBe(0);
}
/// <summary>
/// A probe sync on a driver whose pump was never started (no platform source, e.g. the
/// DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just