Joseph Doherty
a8d3790b09
BLOCKER: the reconnect replay chose work from the registry entries' sessions, but ONE registry entry holds every probe handle and any probe publish on the new session (a seed in the reopen→replay gap, or the straddle retry) re-stamps it as current — the replay then skipped the probe watcher and the old platforms stayed bound to handle 0 (host status frozen); a publish after the snapshot made it re-subscribe everything (double advise, leaked handle). ReplayAsync now excludes the probe entry from its snapshot and calls the new PerPlatformProbeWatcher.ResubscribeStaleAsync on every reconnect, which re-subscribes exactly the platforms whose ProbeHandle.Session predates the current session. The replay-gap path keeps ResubscribeAsync (all platforms). Also: ClearRegistry (teardown) starts a new registry session, so a late PurgeParked from a teardown-orphaned call cannot reach the next runtime's parked values; the read path's unused pump local is gone (call kept). Tests go through the driver's reopen/replay: the probe old-session-number test is rewritten that way, plus the seed-in-the-gap scenario (only P and Q are re-subscribed, R keeps its new-session handle, all three resolve). Both checked red with the probe replay removed / re-subscribing all platforms. Refs #859, #861. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
1035 lines
53 KiB
C#
1035 lines
53 KiB
C#
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.Health;
|
|
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 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 GalaxyDriverOptions Opts() => new(
|
|
new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
|
|
new GalaxyMxAccessOptions("OtOpcUa-A"),
|
|
new GalaxyRepositoryOptions(),
|
|
new GalaxyReconnectOptions());
|
|
|
|
// ===== 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();
|
|
var epoch = registry.CurrentSession;
|
|
|
|
registry.ResolveOrPark(DataChange(7, 1), epoch, out _, 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 epoch = registry.CurrentSession;
|
|
var woken = 0;
|
|
registry.ParkedEventsReady += () => woken++;
|
|
|
|
using (registry.BeginPendingSubscribe())
|
|
{
|
|
registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.ResolveOrPark(DataChange(7, 2), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.ResolveOrPark(DataChange(8, 9), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.TakeReadyEvents(epoch).ShouldBeEmpty();
|
|
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
}
|
|
|
|
woken.ShouldBe(1);
|
|
registry.TakeReadyEvents(epoch).Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]);
|
|
registry.TakeReadyEvents(epoch).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();
|
|
var epoch = registry.CurrentSession;
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
|
|
using (registry.BeginPendingSubscribe())
|
|
{
|
|
registry.ResolveOrPark(DataChange(70, 5), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.Rebind(1, [new TagBinding("Tank.Level", 70)]);
|
|
}
|
|
|
|
registry.TakeReadyEvents(epoch).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 epoch = registry.CurrentSession;
|
|
var first = registry.BeginPendingSubscribe();
|
|
var second = registry.BeginPendingSubscribe();
|
|
|
|
registry.ResolveOrPark(DataChange(7, 1), epoch, out _, 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), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Stale);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #6: a window opening right after the previous one closed never inherits its
|
|
/// leftovers (the N→0 discard and the 0→1 open are serialised under the park gate).
|
|
/// </summary>
|
|
[Fact]
|
|
public void A_new_window_never_inherits_the_previous_windows_leftovers()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var epoch = registry.CurrentSession;
|
|
var first = registry.BeginPendingSubscribe();
|
|
registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
first.Dispose();
|
|
|
|
using var second = registry.BeginPendingSubscribe();
|
|
registry.ParkedEventCount.ShouldBe(0);
|
|
registry.Register(1, [new TagBinding("Reused", 7)]);
|
|
registry.TakeReadyEvents(epoch).ShouldBeEmpty("the closed window's event must not reach the new owner of handle 7");
|
|
}
|
|
|
|
/// <summary>Parking is bounded; past the bound the event is dropped as overflow.</summary>
|
|
[Fact]
|
|
public void Parking_is_bounded()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var epoch = registry.CurrentSession;
|
|
using var window = registry.BeginPendingSubscribe();
|
|
var ev = DataChange(7, 1);
|
|
for (var i = 0; i < SubscriptionRegistry.MaxParkedEvents; i++)
|
|
{
|
|
registry.ResolveOrPark(ev, epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
}
|
|
|
|
registry.ResolveOrPark(ev, epoch, out _, out _).ShouldBe(EarlyEventOutcome.Overflow);
|
|
registry.ParkedEventCount.ShouldBe(SubscriptionRegistry.MaxParkedEvents);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #5 + M-2: promoted-but-not-yet-replayed events stay counted until the dispatch thread
|
|
/// takes them, and the next ResolveOrPark hands them back (in order) before anything else.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Ready_events_stay_counted_until_handed_back_by_the_next_resolve()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var session = registry.CurrentSession;
|
|
using var window = registry.BeginPendingSubscribe();
|
|
registry.ResolveOrPark(DataChange(7, 1), session, out _, out _);
|
|
registry.ResolveOrPark(DataChange(7, 2), session, out _, out _);
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]); // promoted to ready
|
|
registry.ParkedEventCount.ShouldBe(2);
|
|
|
|
registry.ResolveOrPark(DataChange(8, 3), session, out _, out var replayFirst).ShouldBe(EarlyEventOutcome.Parked);
|
|
replayFirst.Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]);
|
|
registry.ParkedEventCount.ShouldBe(1, "only the newly parked event for handle 8 remains held");
|
|
}
|
|
|
|
/// <summary>
|
|
/// I-2: windows that overlap so the open count never reaches 0 (concurrent reads) must not
|
|
/// accumulate unclaimed events — each is trimmed once every window that could claim it closed.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Overlapping_windows_that_never_all_close_do_not_accumulate_unclaimed_events()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var session = registry.CurrentSession;
|
|
var open = registry.BeginPendingSubscribe();
|
|
for (var i = 0; i < 1_000; i++)
|
|
{
|
|
// An unknown-handle event nobody will ever claim (e.g. a trailing event after a read's Remove).
|
|
registry.ResolveOrPark(DataChange(10_000 + i, i), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
var next = registry.BeginPendingSubscribe(); // overlap: the count never drops to 0
|
|
open.Dispose();
|
|
open = next;
|
|
registry.ParkedEventCount.ShouldBe(0, $"iteration {i}: the event's only possible owners have all closed");
|
|
}
|
|
|
|
// Positive control: an event parked inside the still-open window survives and is claimable.
|
|
registry.ResolveOrPark(DataChange(7, 42), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
var later = registry.BeginPendingSubscribe();
|
|
later.Dispose(); // a window opened AFTER the park closing must not trim it
|
|
registry.ParkedEventCount.ShouldBe(1);
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
registry.TakeReadyEvents(session).Single().Value.DoubleValue.ShouldBe(42.0);
|
|
open.Dispose();
|
|
}
|
|
|
|
/// <summary>I-3 interim: an unsubscribe purges events parked for its handles.</summary>
|
|
[Fact]
|
|
public void Purge_drops_events_parked_for_unsubscribed_handles()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var session = registry.CurrentSession;
|
|
using var window = registry.BeginPendingSubscribe();
|
|
registry.ResolveOrPark(DataChange(7, 1), session, out _, out _); // trailing event of a removed item
|
|
registry.ResolveOrPark(DataChange(8, 2), session, out _, out _); // positive control
|
|
|
|
registry.PurgeParked([7], session);
|
|
|
|
registry.ParkedEventCount.ShouldBe(1);
|
|
registry.Register(1, [new TagBinding("Reused", 7), new TagBinding("Other", 8)]);
|
|
registry.TakeReadyEvents(session).Select(e => e.ItemHandle).ShouldBe([8]);
|
|
}
|
|
|
|
/// <summary>
|
|
/// M-1: a live event for a RESOLVED handle that (defensively) still has a parked list is never
|
|
/// parked behind it — the list is promoted and the event resolves, so no trim can lose it.
|
|
/// </summary>
|
|
[Fact]
|
|
public void A_resolved_handle_with_a_stranded_parked_list_promotes_it_and_resolves()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var session = registry.CurrentSession;
|
|
var window = registry.BeginPendingSubscribe();
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
registry.ForceParkForTests(DataChange(7, 1)); // the "should be impossible" stranded list
|
|
|
|
registry.ResolveOrPark(DataChange(7, 2), session, out var subs, out var replayFirst)
|
|
.ShouldBe(EarlyEventOutcome.Resolved);
|
|
subs.Single().FullReference.ShouldBe("Tank.Level");
|
|
replayFirst.Select(e => e.Value.DoubleValue).ShouldBe([1.0]);
|
|
window.Dispose();
|
|
registry.ParkedEventCount.ShouldBe(0);
|
|
}
|
|
|
|
/// <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();
|
|
var epoch = registry.CurrentSession;
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
using var window = registry.BeginPendingSubscribe();
|
|
|
|
registry.ResolveOrPark(DataChange(7, 1), epoch, out var subs, out _).ShouldBe(EarlyEventOutcome.Resolved);
|
|
subs.Single().FullReference.ShouldBe("Tank.Level");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #3: held events belong to a session epoch. A new epoch (a new pump after a
|
|
/// reconnect) discards parked AND promoted-but-unreplayed events, and an older pump can
|
|
/// neither park nor replay — so a reused handle number never receives an old-session value.
|
|
/// </summary>
|
|
[Fact]
|
|
public void A_new_session_discards_held_events_and_fences_the_old_session()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var oldSession = registry.CurrentSession;
|
|
using var window = registry.BeginPendingSubscribe();
|
|
registry.ResolveOrPark(DataChange(7, 111), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.ResolveOrPark(DataChange(8, 112), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.Register(1, [new TagBinding("Old.A", 8)]); // 8 promoted to ready, never replayed
|
|
|
|
var newSession = registry.StartNewSession();
|
|
registry.ParkedEventCount.ShouldBe(0);
|
|
registry.TakeReadyEvents(newSession).ShouldBeEmpty();
|
|
registry.Register(2, [new TagBinding("New.B", 7)]); // new session reuses handle 7
|
|
registry.TakeReadyEvents(newSession).ShouldBeEmpty("the old session's value must not reach New.B");
|
|
|
|
registry.ResolveOrPark(DataChange(7, 1), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Stale,
|
|
"a dying old-session pump must not resolve through the map, even for a registered number");
|
|
registry.ResolveOrPark(DataChange(9, 1), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Stale);
|
|
registry.TakeReadyEvents(oldSession).ShouldBeEmpty();
|
|
registry.ResolveOrPark(DataChange(9, 1), newSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked,
|
|
"positive control: the current session still parks");
|
|
}
|
|
|
|
/// <summary>
|
|
/// I-1: a new session invalidates the reverse map but keeps the subscriptions, so a new-session
|
|
/// event on a number an old binding still holds PARKS (instead of resolving to the old tag) and
|
|
/// is promoted by the replay's Rebind to the right tag.
|
|
/// </summary>
|
|
[Fact]
|
|
public void A_new_session_invalidates_old_handles_so_colliding_numbers_park_until_rebind()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
registry.Register(1, [new TagBinding("Old.A", 7)]);
|
|
registry.Register(2, [new TagBinding("Old.B", 8)]);
|
|
|
|
var session = registry.StartNewSession();
|
|
registry.TrackedSubscriptionCount.ShouldBe(2, "subscriptions survive for the replay to rebind");
|
|
using var window = registry.BeginPendingSubscribe();
|
|
// New session: Old.B is re-issued as handle 7 — the number Old.A still holds until its own Rebind.
|
|
registry.ResolveOrPark(DataChange(7, 5), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
registry.TryRebind(2, [new TagBinding("Old.B", 7)], session).ShouldBe(SubscriptionRegistry.BindOutcome.Bound);
|
|
|
|
registry.TakeReadyEvents(session).ShouldHaveSingleItem().Value.DoubleValue.ShouldBe(5.0);
|
|
registry.ResolveSubscribers(7).ShouldBe([(2L, "Old.B")]);
|
|
}
|
|
|
|
/// <summary>#861: a registration or rebind carrying a stale session is refused.</summary>
|
|
[Fact]
|
|
public void Registration_with_a_stale_session_is_refused()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var before = registry.CurrentSession;
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
registry.StartNewSession();
|
|
|
|
registry.TryRegister(2, [new TagBinding("Late", 9)], before).ShouldBeFalse();
|
|
registry.Contains(2).ShouldBeFalse();
|
|
registry.TryRebind(1, [new TagBinding("Tank.Level", 9)], before).ShouldBe(SubscriptionRegistry.BindOutcome.SessionChanged);
|
|
registry.ResolveSubscribers(9).ShouldBeEmpty();
|
|
registry.TryRegister(3, [new TagBinding("Fresh", 9)], registry.CurrentSession).ShouldBeTrue("positive control");
|
|
|
|
registry.Remove(1, out var live).ShouldNotBeNull();
|
|
live.ShouldBeFalse("never rebound in the new session — its handle 7 must not be unsubscribed");
|
|
registry.Remove(3, out live).ShouldNotBeNull();
|
|
live.ShouldBeTrue();
|
|
}
|
|
|
|
// ===== 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 void Emit(MxEvent ev) => _events.Writer.TryWrite(ev).ShouldBeTrue();
|
|
}
|
|
|
|
/// <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())
|
|
{
|
|
subscriber.Emit(DataChange(7, 42));
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the early event must be parked, not dropped");
|
|
delivered.ShouldBeEmpty();
|
|
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
}
|
|
|
|
await TestWait.UntilAsync(() => !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())
|
|
{
|
|
subscriber.Emit(DataChange(7, 1));
|
|
subscriber.Emit(DataChange(7, 2));
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 2, "both early events must be parked");
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
subscriber.Emit(DataChange(7, 3));
|
|
}
|
|
|
|
await TestWait.UntilAsync(() => delivered.Count >= 3, "all three values must be delivered",
|
|
() => string.Join(",", delivered));
|
|
delivered.ShouldBe([1.0, 2.0, 3.0]);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #2: a live event arriving while a registration has written the reverse map but not
|
|
/// yet promoted (a 50k-binding Register is milliseconds of work) must NOT overtake the parked
|
|
/// initial value. The hook runs inside that window, emits the newer value and gives the pump
|
|
/// ample time to (wrongly) dispatch it; the pump must instead wait for the promotion.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Pump_does_not_let_a_live_event_overtake_a_registration_in_progress()
|
|
{
|
|
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();
|
|
|
|
var deliveredDuringRegistration = -1;
|
|
using (registry.BeginPendingSubscribe())
|
|
{
|
|
subscriber.Emit(DataChange(7, 1));
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the initial value must be parked");
|
|
|
|
registry.AfterMapWriteForTests = () =>
|
|
{
|
|
subscriber.Emit(DataChange(7, 2));
|
|
Thread.Sleep(300);
|
|
deliveredDuringRegistration = delivered.Count;
|
|
};
|
|
registry.Register(1, [new TagBinding("Tank.Level", 7)]);
|
|
registry.AfterMapWriteForTests = null;
|
|
}
|
|
|
|
await TestWait.UntilAsync(() => delivered.Count >= 2, "both values must be delivered",
|
|
() => string.Join(",", delivered));
|
|
deliveredDuringRegistration.ShouldBe(0, "nothing may dispatch for the handle mid-registration");
|
|
delivered.ShouldBe([1.0, 2.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();
|
|
|
|
subscriber.Emit(DataChange(99, 5)); // unsubscribed handle
|
|
subscriber.Emit(DataChange(7, 6)); // positive control, strictly after
|
|
|
|
await TestWait.UntilAsync(() => !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.ParkedEventCount.ShouldBe(0);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #3, pump level: an event the old session's pump parked is never replayed by the
|
|
/// next session's pump onto the tag that now owns the reused handle number.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Pump_after_reconnect_never_replays_an_old_session_event_onto_a_reused_handle()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
|
using var window = registry.BeginPendingSubscribe(); // e.g. a subscribe hanging across the reconnect
|
|
|
|
var oldStream = new ChannelSubscriber();
|
|
var oldPump = new EventPump(oldStream, registry, channelCapacity: 16, clientName: "Old");
|
|
oldPump.Start();
|
|
oldStream.Emit(DataChange(7, 111));
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the old session's event must be parked");
|
|
await oldPump.DisposeAsync();
|
|
registry.StartNewSession(); // what GalaxyDriver.ReplayAsync does before recreating the pump
|
|
|
|
var newStream = new ChannelSubscriber();
|
|
await using var newPump = new EventPump(newStream, registry, channelCapacity: 16, clientName: "New");
|
|
newPump.OnDataChange += (_, e) => delivered.Enqueue(e);
|
|
newPump.Start();
|
|
|
|
registry.Register(2, [new TagBinding("New.Tag", 7)]); // new session reuses handle 7
|
|
newStream.Emit(DataChange(7, 222)); // positive control on the new session
|
|
|
|
await TestWait.UntilAsync(() => !delivered.IsEmpty, "the new session's value must arrive");
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 0, "nothing may stay held");
|
|
delivered.Select(e => (double)e.Snapshot.Value!).ShouldBe([222.0]);
|
|
}
|
|
|
|
// ===== GalaxyDriver / probe watcher 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 consumer's pump has parked that event — the
|
|
/// event deterministically precedes the reply.
|
|
/// </summary>
|
|
private sealed class EventBeforeReplySubscriber : IGalaxySubscriber
|
|
{
|
|
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
|
private int _nextHandle = 100;
|
|
|
|
/// <summary>Gets or sets the consumer's held-event count, polled before replying.</summary>
|
|
public Func<int> ParkedCount { get; set; } = () => 0;
|
|
|
|
/// <summary>Gets the handles unsubscribed so far.</summary>
|
|
public ConcurrentQueue<int> Unsubscribed { get; } = new();
|
|
|
|
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 TestWait.UntilAsync(() => ParkedCount() >= before + fullReferences.Count,
|
|
"the initial values must reach the pump (and be parked) before the reply is returned",
|
|
() => $"held={ParkedCount()} before={before}");
|
|
return results;
|
|
}
|
|
|
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
|
|
{
|
|
foreach (var h in itemHandles) Unsubscribed.Enqueue(h);
|
|
return 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()
|
|
{
|
|
var subscriber = new EventBeforeReplySubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
subscriber.ParkedCount = () => driver.ParkedEarlyEventCountForTests;
|
|
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 TestWait.UntilAsync(() => 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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Review #1: the subscribe-once read waits ONLY for the initial value, so an initial value
|
|
/// that overtakes the reply must still complete the read Good — not BadTimeout.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Driver_read_returns_an_initial_value_that_arrived_before_the_subscribe_reply()
|
|
{
|
|
var subscriber = new EventBeforeReplySubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
subscriber.ParkedCount = () => driver.ParkedEarlyEventCountForTests;
|
|
|
|
var snapshots = await TestWait.ForAsync(
|
|
driver.ReadAsync(["Tank.Level"], TestContext.Current.CancellationToken),
|
|
"the read must complete");
|
|
|
|
var snapshot = snapshots.ShouldHaveSingleItem();
|
|
snapshot.StatusCode.ShouldBe(StatusCodeMap.Good);
|
|
((double)snapshot.Value!).ShouldBe(1010.0);
|
|
driver.ParkedEarlyEventCountForTests.ShouldBe(0);
|
|
}
|
|
|
|
/// <summary>The probe watcher's initial ScanState overtaking its reply still reaches the fan-out.</summary>
|
|
[Fact]
|
|
public async Task Probe_watcher_initial_value_that_arrived_before_the_reply_is_dispatched()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var subscriber = new EventBeforeReplySubscriber { ParkedCount = () => registry.ParkedEventCount };
|
|
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
|
await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Probe");
|
|
pump.OnDataChange += (_, e) => delivered.Enqueue(e);
|
|
pump.Start();
|
|
using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry);
|
|
|
|
await watcher.SyncPlatformsAsync(["PlatformA"], TestContext.Current.CancellationToken);
|
|
|
|
await TestWait.UntilAsync(() => !delivered.IsEmpty, "the initial ScanState must be dispatched");
|
|
var args = delivered.Single();
|
|
args.FullReference.ShouldBe("PlatformA.ScanState");
|
|
((GalaxySubscriptionHandle)args.SubscriptionHandle).SubscriptionId.ShouldBe(SubscriptionRegistry.ProbeSubscriptionId);
|
|
}
|
|
|
|
// ===== Reconnect / session (#859 I-1, #861) =====
|
|
|
|
/// <summary>
|
|
/// A gateway fake whose SubscribeBulk calls follow a per-call script (default: fresh handles).
|
|
/// Events go on one stream that survives pump restarts.
|
|
/// </summary>
|
|
private sealed class ScriptedSubscriber : IGalaxySubscriber
|
|
{
|
|
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
|
private int _nextHandle = 1_000;
|
|
|
|
public ConcurrentQueue<Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<SubscribeResult>>>> Script { get; } = new();
|
|
public ConcurrentQueue<int> Unsubscribed { get; } = new();
|
|
public int SubscribeCalls;
|
|
|
|
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
|
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
|
{
|
|
Interlocked.Increment(ref SubscribeCalls);
|
|
if (Script.TryDequeue(out var step)) return step(fullReferences, cancellationToken);
|
|
return Task.FromResult<IReadOnlyList<SubscribeResult>>(
|
|
[.. fullReferences.Select(r => Ok(r, Interlocked.Increment(ref _nextHandle)))]);
|
|
}
|
|
|
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
|
|
{
|
|
foreach (var h in itemHandles) Unsubscribed.Enqueue(h);
|
|
return 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 void Emit(MxEvent ev) => _events.Writer.TryWrite(ev).ShouldBeTrue();
|
|
}
|
|
|
|
private static SubscribeResult Ok(string tag, int handle) =>
|
|
new() { TagAddress = tag, ItemHandle = handle, WasSuccessful = true };
|
|
|
|
/// <summary>
|
|
/// I-1 end to end: after a reconnect the new session hands Tank.A the number Tank.B held in the
|
|
/// old session (and vice versa), and each initial value arrives BEFORE its SubscribeBulk reply.
|
|
/// Every value must reach its own tag — not the tag that owned the number before the reconnect.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Replay_with_colliding_handles_and_event_before_reply_delivers_each_value_to_its_own_tag()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
var registry = driver.SubscriptionsForTests;
|
|
var delivered = new ConcurrentQueue<(string Ref, double Value)>();
|
|
driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!));
|
|
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 1)]));
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 2)]));
|
|
await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
await driver.SubscribeAsync(["Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
|
|
// New session: swapped numbers; each initial value is on the stream (and parked) before the reply.
|
|
var newHandle = new Dictionary<string, int> { ["Tank.A"] = 2, ["Tank.B"] = 1 };
|
|
var newValue = new Dictionary<string, double> { ["Tank.A"] = 10.0, ["Tank.B"] = 20.0 };
|
|
for (var i = 0; i < 2; i++)
|
|
{
|
|
subscriber.Script.Enqueue(async (refs, _) =>
|
|
{
|
|
var tag = refs[0];
|
|
// The previous iteration's promoted value stays counted until the pump replays it.
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount == 0, "earlier promoted values must be replayed first");
|
|
subscriber.Emit(DataChange(newHandle[tag], newValue[tag]));
|
|
await TestWait.UntilAsync(() => registry.ParkedEventCount > 0,
|
|
$"{tag}'s new-session initial value must PARK, not resolve to the old owner of handle {newHandle[tag]}");
|
|
return [Ok(tag, newHandle[tag])];
|
|
});
|
|
}
|
|
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
await TestWait.UntilAsync(() => delivered.Count >= 2, "both replayed initial values must be published",
|
|
() => string.Join(", ", delivered));
|
|
delivered.OrderBy(d => d.Ref).ShouldBe([("Tank.A", 10.0), ("Tank.B", 20.0)]);
|
|
registry.ResolveSubscribers(1).Single().FullReference.ShouldBe("Tank.B");
|
|
registry.ResolveSubscribers(2).Single().FullReference.ShouldBe("Tank.A");
|
|
}
|
|
|
|
/// <summary>
|
|
/// #861: a SubscribeAsync whose reply returns after a reconnect does not register the dead
|
|
/// old-session handles (nor unsubscribe their numbers) — it re-issues on the new session.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Subscribe_whose_reply_straddles_a_reconnect_is_reissued_on_the_new_session()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
var registry = driver.SubscriptionsForTests;
|
|
var replyGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
subscriber.Script.Enqueue(async (refs, _) =>
|
|
{
|
|
await replyGate.Task;
|
|
return [Ok(refs[0], 500)]; // an old-session handle
|
|
});
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 600)]));
|
|
|
|
var subscribe = driver.SubscribeAsync(["Tank.X"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the first SubscribeBulk must be in flight");
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); // the reconnect
|
|
replyGate.SetResult();
|
|
|
|
var handle = await TestWait.ForAsync(subscribe, "the subscribe must complete on the new session");
|
|
Volatile.Read(ref subscriber.SubscribeCalls).ShouldBe(2);
|
|
registry.ResolveSubscribers(500).ShouldBeEmpty("the dead old-session handle must not be registered");
|
|
registry.ResolveSubscribers(600).ShouldBe([(((GalaxySubscriptionHandle)handle).SubscriptionId, "Tank.X")]);
|
|
subscriber.Unsubscribed.ShouldNotContain(500, "its number may belong to a new-session item");
|
|
}
|
|
|
|
/// <summary>#861: a subscribe-once read whose reply straddles a reconnect reports the lost session.</summary>
|
|
[Fact]
|
|
public async Task Read_whose_reply_straddles_a_reconnect_returns_bad_communication_and_leaves_old_handles_alone()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
var registry = driver.SubscriptionsForTests;
|
|
var replyGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
subscriber.Script.Enqueue(async (refs, _) =>
|
|
{
|
|
await replyGate.Task;
|
|
return [Ok(refs[0], 700)];
|
|
});
|
|
|
|
var read = driver.ReadAsync(["Tank.Y"], TestContext.Current.CancellationToken);
|
|
await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the read's SubscribeBulk must be in flight");
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
replyGate.SetResult();
|
|
|
|
var snapshot = (await TestWait.ForAsync(read, "the read must complete")).ShouldHaveSingleItem();
|
|
snapshot.StatusCode.ShouldBe(StatusCodeMap.BadCommunicationError);
|
|
registry.ResolveSubscribers(700).ShouldBeEmpty();
|
|
subscriber.Unsubscribed.ShouldBeEmpty();
|
|
registry.PendingSubscribeCount.ShouldBe(0, "the read's early-event window must be closed");
|
|
}
|
|
|
|
/// <summary>The read's internal-deadline path closes its early-event window.</summary>
|
|
[Fact]
|
|
public async Task Read_deadline_path_closes_its_window()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
|
|
{
|
|
ReadViaSubscribeDeadline = TimeSpan.FromMilliseconds(100),
|
|
};
|
|
var never = new TaskCompletionSource<IReadOnlyList<SubscribeResult>>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
subscriber.Script.Enqueue((_, _) => never.Task);
|
|
|
|
var snapshot = (await TestWait.ForAsync(
|
|
driver.ReadAsync(["Tank.Z"], TestContext.Current.CancellationToken), "the read must time out")).ShouldHaveSingleItem();
|
|
|
|
snapshot.StatusCode.ShouldBe(StatusCodeMap.BadTimeout);
|
|
driver.SubscriptionsForTests.PendingSubscribeCount.ShouldBe(0);
|
|
never.SetResult([]); // let the abandoned RPC finish cleanly
|
|
}
|
|
|
|
// ===== Third review (B-1, I-A, I-B, I-C, P-1, M-a) =====
|
|
|
|
/// <summary>
|
|
/// B-1: a read rebound by a reconnect replay while it waited unsubscribes its NEW handle, not the
|
|
/// original number (which may by now be another tag's live item).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Read_rebound_by_a_replay_unsubscribes_the_new_handle_not_the_original()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 1)]));
|
|
subscriber.Script.Enqueue((refs, _) =>
|
|
{
|
|
subscriber.Emit(DataChange(9, 99.0)); // the replayed initial value (may overtake the reply)
|
|
return Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 9)]);
|
|
});
|
|
|
|
var read = driver.ReadAsync(["Tank.R"], TestContext.Current.CancellationToken);
|
|
await TestWait.UntilAsync(() => driver.TrackedSubscriptionCountForTests == 1, "the read must be registered and waiting");
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
var snapshot = (await TestWait.ForAsync(read, "the replayed value must complete the read")).ShouldHaveSingleItem();
|
|
((double)snapshot.Value!).ShouldBe(99.0);
|
|
subscriber.Unsubscribed.ToArray().ShouldBe([9], "the read must release its NEW handle, never the original number 1");
|
|
}
|
|
|
|
/// <summary>P-1: a reconnect with nothing to replay still replaces the dead pump; a later subscribe gets data.</summary>
|
|
[Fact]
|
|
public async Task Reconnect_with_no_subscriptions_restarts_the_pump()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
var delivered = new ConcurrentQueue<DataChangeEventArgs>();
|
|
driver.OnDataChange += (_, e) => delivered.Enqueue(e);
|
|
var first = await driver.SubscribeAsync(["Tank.Old"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
await driver.UnsubscribeAsync(first, TestContext.Current.CancellationToken);
|
|
var deadPump = driver.EventPumpInstanceForTests.ShouldNotBeNull();
|
|
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
driver.EventPumpInstanceForTests.ShouldNotBeSameAs(deadPump, "the faulted pump must be replaced even with no entries");
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 42)]));
|
|
await driver.SubscribeAsync(["Tank.New"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
subscriber.Emit(DataChange(42, 4.2));
|
|
await TestWait.UntilAsync(() => !delivered.IsEmpty, "a subscribe after the reconnect must receive data");
|
|
delivered.Single().FullReference.ShouldBe("Tank.New");
|
|
}
|
|
|
|
/// <summary>
|
|
/// I-C: the reopen fences the registry around the session swap. An unsubscribe during the
|
|
/// reopen/backoff never sends an old number; a subscribe issued after the swap registers normally
|
|
/// and the replay does not re-subscribe it (no double advise).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Subscribe_after_the_session_swap_but_before_replay_is_live_and_not_replayed()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
|
|
{
|
|
RecreateOverrideForTests = _ => Task.CompletedTask,
|
|
};
|
|
var delivered = new ConcurrentQueue<(string Ref, double Value)>();
|
|
driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!));
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 1)]));
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 3)]));
|
|
await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
var c = await driver.SubscribeAsync(["Tank.C"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
|
|
await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken); // gateway session swapped
|
|
|
|
await driver.UnsubscribeAsync(c, TestContext.Current.CancellationToken);
|
|
subscriber.Unsubscribed.ShouldNotContain(3, "an old-session number must never reach the new session");
|
|
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 50)]));
|
|
await driver.SubscribeAsync(["Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
var callsBeforeReplay = Volatile.Read(ref subscriber.SubscribeCalls);
|
|
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 51)]));
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
(Volatile.Read(ref subscriber.SubscribeCalls) - callsBeforeReplay).ShouldBe(1, "only Tank.A (old session) is re-subscribed");
|
|
var registry = driver.SubscriptionsForTests;
|
|
registry.ResolveSubscribers(50).Single().FullReference.ShouldBe("Tank.B");
|
|
registry.ResolveSubscribers(51).Single().FullReference.ShouldBe("Tank.A");
|
|
subscriber.Emit(DataChange(50, 5.0));
|
|
await TestWait.UntilAsync(() => !delivered.IsEmpty, "the post-swap subscription must stay live across the replay");
|
|
delivered.Single().ShouldBe(("Tank.B", 5.0));
|
|
}
|
|
|
|
/// <summary>
|
|
/// M-a: a subscription removed while the replay's SubscribeBulk for it was in flight gets the
|
|
/// freshly advised handles released (nothing else would ever unsubscribe them).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Replay_releases_handles_for_a_subscription_removed_mid_replay()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 1)]));
|
|
var handle = await driver.SubscribeAsync(["Tank.Gone"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
|
|
subscriber.Script.Enqueue(async (refs, _) =>
|
|
{
|
|
await driver.UnsubscribeAsync(handle, CancellationToken.None); // removed mid-replay
|
|
return [Ok(refs[0], 77)];
|
|
});
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
subscriber.Unsubscribed.ToArray().ShouldBe([77], "the re-advised handle is released; the dead number 1 is not sent");
|
|
driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty();
|
|
}
|
|
|
|
/// <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()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var oldSession = registry.CurrentSession;
|
|
var newSession = registry.StartNewSession();
|
|
using var window = registry.BeginPendingSubscribe();
|
|
registry.ResolveOrPark(DataChange(7, 1), newSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked);
|
|
|
|
registry.PurgeParked([7], oldSession); // a slow old-session unsubscribe finishing late
|
|
registry.ParkedEventCount.ShouldBe(1, "the new session's parked initial value must survive");
|
|
|
|
registry.PurgeParked([7], newSession); // positive control
|
|
registry.ParkedEventCount.ShouldBe(0);
|
|
}
|
|
|
|
/// <summary>I-A: a probe seed whose reply straddles a reconnect records nothing dead and is re-issued.</summary>
|
|
[Fact]
|
|
public async Task Probe_seed_straddling_a_reconnect_is_reissued_and_never_publishes_the_dead_handle()
|
|
{
|
|
var registry = new SubscriptionRegistry();
|
|
var subscriber = new ScriptedSubscriber();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
subscriber.Script.Enqueue(async (refs, _) =>
|
|
{
|
|
await gate.Task;
|
|
return [Ok(refs[0], 5)]; // old-session handle
|
|
});
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 6)]));
|
|
using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry);
|
|
|
|
var sync = watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken);
|
|
await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the seed's SubscribeBulk must be in flight");
|
|
registry.StartNewSession();
|
|
gate.SetResult();
|
|
await TestWait.ForAsync(sync, "the seed must complete");
|
|
|
|
Volatile.Read(ref subscriber.SubscribeCalls).ShouldBe(2);
|
|
watcher.CurrentBindings.Single().ItemHandle.ShouldBe(6);
|
|
registry.ResolveSubscribers(5).ShouldBeEmpty();
|
|
registry.ResolveSubscribers(6).Single().FullReference.ShouldBe("P.ScanState");
|
|
subscriber.Unsubscribed.ShouldBeEmpty();
|
|
}
|
|
|
|
/// <summary>
|
|
/// I-A: after a reconnect (before the replay rebinds probes), a seed neither re-publishes the old
|
|
/// session's probe numbers nor unsubscribes a removed platform's old number.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Probe_watcher_never_publishes_or_unsubscribes_old_session_numbers()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = await InitializedDriverAsync(subscriber);
|
|
var watcher = driver.ProbeWatcherForTests.ShouldNotBeNull();
|
|
var registry = driver.SubscriptionsForTests;
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>(
|
|
[.. refs.Select((r, i) => Ok(r, i + 1))])); // P → 1, Q → 2
|
|
await watcher.SyncPlatformsAsync(["P", "Q"], TestContext.Current.CancellationToken);
|
|
|
|
await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken); // gateway session swapped
|
|
// The new session reuses number 1 for a user tag.
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 1)]));
|
|
await driver.SubscribeAsync(["User.Tag"], TimeSpan.Zero, TestContext.Current.CancellationToken);
|
|
|
|
await watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken); // a seed in the gap drops Q
|
|
|
|
subscriber.Unsubscribed.ShouldBeEmpty("Q's handle 2 is an old-session number");
|
|
registry.ResolveSubscribers(1).Single().FullReference.ShouldBe("User.Tag", "P's old-session number must not be re-published");
|
|
|
|
var replayed = new ConcurrentQueue<string>();
|
|
subscriber.Script.Enqueue((refs, _) =>
|
|
{
|
|
foreach (var r in refs) replayed.Enqueue(r);
|
|
return Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 7)]);
|
|
});
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); // the driver's replay
|
|
|
|
replayed.ToArray().ShouldBe(["P.ScanState"]);
|
|
watcher.CurrentBindings.Single().ItemHandle.ShouldBe(7);
|
|
registry.ResolveSubscribers(7).Single().FullReference.ShouldBe("P.ScanState");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Final review blocker: a probe seed in the gap between the session swap and the replay
|
|
/// re-stamps the shared probe registry entry as current. The replay must still re-subscribe the
|
|
/// platforms whose handles predate the swap (they would otherwise freeze at handle 0) — and only
|
|
/// those: the platform the seed just advised on the new session is not advised twice.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Replay_after_a_seed_in_the_gap_resubscribes_exactly_the_stale_platforms()
|
|
{
|
|
var subscriber = new ScriptedSubscriber();
|
|
using var driver = await InitializedDriverAsync(subscriber);
|
|
var watcher = driver.ProbeWatcherForTests.ShouldNotBeNull();
|
|
var registry = driver.SubscriptionsForTests;
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>(
|
|
[.. refs.Select((r, i) => Ok(r, i + 1))])); // P → 1, Q → 2
|
|
await watcher.SyncPlatformsAsync(["P", "Q"], TestContext.Current.CancellationToken);
|
|
|
|
await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken);
|
|
subscriber.Script.Enqueue((refs, _) => Task.FromResult<IReadOnlyList<SubscribeResult>>([Ok(refs[0], 20)]));
|
|
await watcher.SyncPlatformsAsync(["P", "Q", "R"], TestContext.Current.CancellationToken); // seed adds R on the new session
|
|
registry.ResolveSubscribers(20).Single().FullReference.ShouldBe("R.ScanState");
|
|
|
|
var replayed = new ConcurrentQueue<string>();
|
|
subscriber.Script.Enqueue((refs, _) =>
|
|
{
|
|
foreach (var r in refs) replayed.Enqueue(r);
|
|
return Task.FromResult<IReadOnlyList<SubscribeResult>>([.. refs.Select((r, i) => Ok(r, 30 + i))]);
|
|
});
|
|
await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
|
|
|
|
replayed.OrderBy(r => r).ToArray().ShouldBe(["P.ScanState", "Q.ScanState"], "R was advised on the new session already");
|
|
var bindings = watcher.CurrentBindings.ToDictionary(b => b.FullReference, b => b.ItemHandle);
|
|
bindings["R.ScanState"].ShouldBe(20);
|
|
bindings["P.ScanState"].ShouldBeGreaterThanOrEqualTo(30);
|
|
bindings["Q.ScanState"].ShouldBeGreaterThanOrEqualTo(30);
|
|
foreach (var (reference, handle) in bindings)
|
|
{
|
|
registry.ResolveSubscribers(handle).Single().FullReference.ShouldBe(reference);
|
|
}
|
|
}
|
|
|
|
private static async Task<GalaxyDriver> InitializedDriverAsync(ScriptedSubscriber subscriber)
|
|
{
|
|
var driver = new GalaxyDriver(
|
|
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
|
|
{
|
|
RecreateOverrideForTests = _ => Task.CompletedTask,
|
|
};
|
|
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
return driver;
|
|
}
|
|
}
|