fix(galaxy): park data-change events that overtake their SubscribeBulk reply (#859)
mxaccessgw gives no reply-before-event ordering: the worker completes the STA
command via a RunContinuationsAsynchronously TCS, so the reply is framed on a
thread-pool continuation while the STA already pumps the OnDataChange Advise
queued (drained/framed by a separate loop), and the reply (unary Invoke) and
the event (StreamEvents) are separate HTTP/2 streams. The driver then also
registers handles concurrently with the pump's dispatch loop. MXAccess's
initial value could therefore be dropped as "stale after unsubscribe",
leaving the node at its pre-subscribe value until the next change.
SubscriptionRegistry gains an early-event window: every SubscribeBulk call
site (SubscribeAsync, ResubscribeAllAsync, PerPlatformProbeWatcher) opens
BeginPendingSubscribe() before the RPC and closes it after registering.
Inside the window unknown-handle events are parked (bounded, 100k); the
Register/Rebind that binds a handle promotes them, and the EventPump replays
them on its single dispatch thread before any later event for that handle
(woken by a sentinel when no further event arrives). Closing the last window
discards unclaimed events, so the stale drop is unchanged outside it.
New counters galaxy.events.early_{parked,replayed,overflow}.
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,315 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading.Channels;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
|
||||
|
||||
/// <summary>
|
||||
/// #859 — MXAccess's initial value for a freshly advised item can reach the event stream
|
||||
/// before the SubscribeBulk reply is registered: mxaccessgw frames the reply on a thread-pool
|
||||
/// continuation while the STA is already pumping the OnDataChange, and the reply and the event
|
||||
/// ride separate RPCs. Such an event used to be dropped as "stale after unsubscribe", leaving
|
||||
/// the node at its pre-subscribe value until the next change. These tests deliver the event
|
||||
/// strictly BEFORE the reply and pin that it is parked and replayed in order, while a genuinely
|
||||
/// stale event (no subscribe in flight) is still dropped.
|
||||
/// </summary>
|
||||
public sealed class EarlyEventWindowTests
|
||||
{
|
||||
private const int WaitMs = 10_000;
|
||||
|
||||
private static MxEvent DataChange(int handle, double value) => new()
|
||||
{
|
||||
Family = MxEventFamily.OnDataChange,
|
||||
ItemHandle = handle,
|
||||
Value = new MxValue { DoubleValue = value },
|
||||
Quality = 192,
|
||||
SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
||||
};
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> predicate, string because)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs);
|
||||
while (!predicate())
|
||||
{
|
||||
if (DateTime.UtcNow > deadline) throw new TimeoutException(because);
|
||||
await Task.Delay(5);
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Registry =====
|
||||
|
||||
/// <summary>With no subscribe in flight an unknown handle is stale — the pre-#859 drop is kept.</summary>
|
||||
[Fact]
|
||||
public void Unknown_handle_without_a_pending_subscribe_is_stale()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
|
||||
registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Stale);
|
||||
registry.ParkedEventCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
/// <summary>Parked events are promoted, in arrival order, when the handle is registered.</summary>
|
||||
[Fact]
|
||||
public void Parked_events_are_promoted_in_order_on_register()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
var woken = 0;
|
||||
registry.ParkedEventsReady += () => woken++;
|
||||
|
||||
using (registry.BeginPendingSubscribe())
|
||||
{
|
||||
registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
registry.ResolveOrPark(DataChange(7, 2), out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
registry.ResolveOrPark(DataChange(8, 9), out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
registry.TakeReadyEvents().ShouldBeEmpty();
|
||||
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
}
|
||||
|
||||
woken.ShouldBe(1);
|
||||
registry.TakeReadyEvents().Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]);
|
||||
registry.TakeReadyEvents().ShouldBeEmpty();
|
||||
// Handle 8 was never registered: closing the last window discarded it.
|
||||
registry.ParkedEventCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
/// <summary>A reconnect/re-snapshot Rebind promotes the new handles' early events too.</summary>
|
||||
[Fact]
|
||||
public void Rebind_promotes_parked_events_for_the_new_handles()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
|
||||
using (registry.BeginPendingSubscribe())
|
||||
{
|
||||
registry.ResolveOrPark(DataChange(70, 5), out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
registry.Rebind(1, [new TagBinding("Tank.Level", 70)]);
|
||||
}
|
||||
|
||||
registry.TakeReadyEvents().Single().ItemHandle.ShouldBe(70);
|
||||
}
|
||||
|
||||
/// <summary>Closing one of two overlapping windows keeps the other subscribe's parked events.</summary>
|
||||
[Fact]
|
||||
public void Closing_one_of_two_windows_keeps_parked_events()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
var first = registry.BeginPendingSubscribe();
|
||||
var second = registry.BeginPendingSubscribe();
|
||||
|
||||
registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
first.Dispose();
|
||||
first.Dispose(); // idempotent — must not close the second window
|
||||
registry.PendingSubscribeCount.ShouldBe(1);
|
||||
registry.ParkedEventCount.ShouldBe(1);
|
||||
|
||||
second.Dispose();
|
||||
registry.ParkedEventCount.ShouldBe(0);
|
||||
registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Stale);
|
||||
}
|
||||
|
||||
/// <summary>Parking is bounded; past the bound the event is dropped as overflow.</summary>
|
||||
[Fact]
|
||||
public void Parking_is_bounded()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
using var window = registry.BeginPendingSubscribe();
|
||||
var ev = DataChange(7, 1);
|
||||
for (var i = 0; i < SubscriptionRegistry.MaxParkedEvents; i++)
|
||||
{
|
||||
registry.ResolveOrPark(ev, out _).ShouldBe(EarlyEventOutcome.Parked);
|
||||
}
|
||||
|
||||
registry.ResolveOrPark(ev, out _).ShouldBe(EarlyEventOutcome.Overflow);
|
||||
registry.ParkedEventCount.ShouldBe(SubscriptionRegistry.MaxParkedEvents);
|
||||
}
|
||||
|
||||
/// <summary>A registered handle resolves normally even while a window is open.</summary>
|
||||
[Fact]
|
||||
public void Registered_handle_resolves_while_a_window_is_open()
|
||||
{
|
||||
var registry = new SubscriptionRegistry();
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
using var window = registry.BeginPendingSubscribe();
|
||||
|
||||
registry.ResolveOrPark(DataChange(7, 1), out var subs).ShouldBe(EarlyEventOutcome.Resolved);
|
||||
subs.Single().FullReference.ShouldBe("Tank.Level");
|
||||
}
|
||||
|
||||
// ===== EventPump =====
|
||||
|
||||
private sealed class ChannelSubscriber : IGalaxySubscriber
|
||||
{
|
||||
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
||||
|
||||
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
||||
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
||||
=> Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
|
||||
|
||||
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return ev;
|
||||
}
|
||||
}
|
||||
|
||||
public ValueTask EmitAsync(MxEvent ev) => _events.Writer.WriteAsync(ev);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An initial value delivered before registration reaches the handler once the handle is
|
||||
/// registered — with NO further event on the stream (the wake path).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Pump_replays_an_early_event_once_the_handle_registers()
|
||||
{
|
||||
var subscriber = new ChannelSubscriber();
|
||||
var registry = new SubscriptionRegistry();
|
||||
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Early");
|
||||
pump.OnDataChange += (_, e) => delivered.Enqueue(e);
|
||||
pump.Start();
|
||||
|
||||
using (registry.BeginPendingSubscribe())
|
||||
{
|
||||
await subscriber.EmitAsync(DataChange(7, 42));
|
||||
await WaitUntilAsync(() => registry.ParkedEventCount == 1, "the early event must be parked, not dropped");
|
||||
delivered.ShouldBeEmpty();
|
||||
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
}
|
||||
|
||||
await WaitUntilAsync(() => !delivered.IsEmpty, "the parked initial value must be replayed");
|
||||
var args = delivered.Single();
|
||||
args.FullReference.ShouldBe("Tank.Level");
|
||||
((double)args.Snapshot.Value!).ShouldBe(42.0);
|
||||
}
|
||||
|
||||
/// <summary>The replayed early value precedes a later live value for the same handle.</summary>
|
||||
[Fact]
|
||||
public async Task Pump_replays_early_events_before_later_live_events()
|
||||
{
|
||||
var subscriber = new ChannelSubscriber();
|
||||
var registry = new SubscriptionRegistry();
|
||||
var delivered = new ConcurrentQueue<double>();
|
||||
await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Early");
|
||||
pump.OnDataChange += (_, e) => delivered.Enqueue((double)e.Snapshot.Value!);
|
||||
pump.Start();
|
||||
|
||||
using (registry.BeginPendingSubscribe())
|
||||
{
|
||||
await subscriber.EmitAsync(DataChange(7, 1));
|
||||
await subscriber.EmitAsync(DataChange(7, 2));
|
||||
await WaitUntilAsync(() => registry.ParkedEventCount == 2, "both early events must be parked");
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
await subscriber.EmitAsync(DataChange(7, 3));
|
||||
}
|
||||
|
||||
await WaitUntilAsync(() => delivered.Count >= 3, "all three values must be delivered");
|
||||
delivered.ShouldBe([1.0, 2.0, 3.0]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A genuinely stale event (no subscribe in flight) is still dropped; the positive control
|
||||
/// that follows it proves the pump processed it rather than not having got to it yet.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Pump_still_drops_a_stale_event_outside_any_window()
|
||||
{
|
||||
var subscriber = new ChannelSubscriber();
|
||||
var registry = new SubscriptionRegistry();
|
||||
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
||||
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Early");
|
||||
pump.OnDataChange += (_, e) => delivered.Enqueue(e);
|
||||
pump.Start();
|
||||
|
||||
await subscriber.EmitAsync(DataChange(99, 5)); // unsubscribed handle
|
||||
await subscriber.EmitAsync(DataChange(7, 6)); // positive control, strictly after
|
||||
|
||||
await WaitUntilAsync(() => !delivered.IsEmpty, "the positive control must arrive");
|
||||
delivered.Single().FullReference.ShouldBe("Tank.Level");
|
||||
registry.ParkedEventCount.ShouldBe(0);
|
||||
|
||||
// Registering 99 now must not resurrect the stale event.
|
||||
registry.Register(2, [new TagBinding("Other", 99)]);
|
||||
registry.TakeReadyEvents().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
// ===== GalaxyDriver end-to-end =====
|
||||
|
||||
/// <summary>
|
||||
/// A gateway fake that writes each new handle's initial value to the event stream and only
|
||||
/// returns the SubscribeBulk reply once the driver's pump has consumed that event — the
|
||||
/// event deterministically precedes the reply.
|
||||
/// </summary>
|
||||
private sealed class EventBeforeReplySubscriber(Func<int> parkedCount) : IGalaxySubscriber
|
||||
{
|
||||
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
||||
private int _nextHandle = 100;
|
||||
|
||||
public async Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
||||
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
||||
{
|
||||
var results = new List<SubscribeResult>();
|
||||
var before = parkedCount();
|
||||
foreach (var fullRef in fullReferences)
|
||||
{
|
||||
var handle = Interlocked.Increment(ref _nextHandle);
|
||||
await _events.Writer.WriteAsync(DataChange(handle, handle * 10.0), cancellationToken);
|
||||
results.Add(new SubscribeResult { TagAddress = fullRef, ItemHandle = handle, WasSuccessful = true });
|
||||
}
|
||||
|
||||
await WaitUntilAsync(() => parkedCount() >= before + fullReferences.Count,
|
||||
"the initial values must reach the pump (and be parked) before the reply is returned");
|
||||
return results;
|
||||
}
|
||||
|
||||
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return ev;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The driver publishes MXAccess's initial value even when it overtook the subscribe reply.</summary>
|
||||
[Fact]
|
||||
public async Task Driver_publishes_an_initial_value_that_arrived_before_the_subscribe_reply()
|
||||
{
|
||||
GalaxyDriver? driverRef = null;
|
||||
var subscriber = new EventBeforeReplySubscriber(() => driverRef!.ParkedEarlyEventCountForTests);
|
||||
using var driver = new GalaxyDriver(
|
||||
"g",
|
||||
new GalaxyDriverOptions(
|
||||
new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
|
||||
new GalaxyMxAccessOptions("OtOpcUa-A"),
|
||||
new GalaxyRepositoryOptions(),
|
||||
new GalaxyReconnectOptions()),
|
||||
hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
||||
driverRef = driver;
|
||||
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
driver.OnDataChange += (_, e) => delivered.Enqueue(e);
|
||||
|
||||
var handle = await driver.SubscribeAsync(
|
||||
["Tank.Level", "Tank.Temp"], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
||||
|
||||
await WaitUntilAsync(() => delivered.Count >= 2, "both initial values must be published");
|
||||
delivered.Select(e => e.FullReference).OrderBy(r => r).ShouldBe(["Tank.Level", "Tank.Temp"]);
|
||||
delivered.ShouldAllBe(e => e.SubscriptionHandle.Equals(handle));
|
||||
driver.ParkedEarlyEventCountForTests.ShouldBe(0);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user