diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index b768aaed8..c7674793e 100644 --- a/docs/drivers/Galaxy.md +++ b/docs/drivers/Galaxy.md @@ -172,6 +172,7 @@ Every (re)open of `StreamAlarms` replays the gateway's active-alarm snapshot. `G - **Replay gaps** (`galaxy.events.replay_gaps`): the gateway signals that events were evicted from its replay ring before we could read them. The driver responds by re-issuing `SubscribeBulk` for every tracked subscription — MXAccess delivers an initial value per item on subscribe, so re-subscribing *is* the re-snapshot the contract asks for. It does **not** restart the event pump (the stream carrying the sentinel is healthy) and it honours `Reconnect:ReplayOnSessionLost=false` by logging and standing down, on the grounds that an operator who delegated recovery to the gateway did not ask for a re-subscribe on a different trigger. ⚠️ This cannot currently fire: the sentinel is only emitted on a stream opened with `after_worker_sequence > 0`, and the driver opens with `0`, which the gateway treats as "fresh stream, not a resume". Enabling resume is a bigger change than it looks — the sequence is only meaningful **within one gateway session**, and the reconnect path recreates the session, so a remembered sequence replayed against a fresh session would set the server's live watermark *above* every event it is about to send, silencing the stream rather than recovering it. - **Undecodable values** (`galaxy.events.undecodable`): a value kind this driver has no CLR representation for is published as **`BadTypeMismatch` with a null value**, and the first occurrence of each distinct kind is logged. It is *not* published as a null carrying the event's own quality: an MXAccess event can be Good-quality while the payload is one we cannot map, and a good null is indistinguishable from a genuine null reading — a client would consume a value it never received as a real one, with the failure visible only in a log. Non-zero here means the gateway is sending a wire type `MxValueDecoder` has no mapping for and those readings are being lost; the fix is a decoder mapping, not a quality change. +- **Initial values can overtake the subscribe reply (#859)** (`galaxy.events.early_parked` / `early_replayed` / `early_overflow`): mxaccessgw gives no reply-before-event ordering — the worker frames the SubscribeBulk reply on a thread-pool continuation while the STA is already pumping the OnDataChange that Advise queued, and the reply (unary Invoke) and the event (StreamEvents) are separate HTTP/2 streams anyway. An event for an unregistered handle is therefore only "stale after unsubscribe" when no subscribe is in flight. Every SubscribeBulk call site (`SubscribeAsync`, the reconnect/replay-gap `ResubscribeAllAsync`, `PerPlatformProbeWatcher`) opens `SubscriptionRegistry.BeginPendingSubscribe()` before the RPC and closes it after registering; inside the window unknown-handle events are parked (bounded at 100 000), promoted by the `Register`/`Rebind` that binds the handle, and replayed on the pump's dispatch thread before any later event for that handle. When the last window closes, unclaimed parked events are discarded — the stale drop is unchanged outside it. - **Duplicate `attributeRef` (#605 / 06-S-13)**: two authored RawPaths dialling the same Galaxy attribute are both dialled, but the reverse map is first-wins, so value fan-out and alarm `ConditionId` surface under the first RawPath only. The loser is logged at Warning and recorded in the skipped-tag ledger with the winner path in the reason. - **Subscribe-once reads are bounded (#605 / 06-S-3)**: `ReadAsync` on the production path (SubscribeBulk, first event, Unsubscribe) carries an internal 30 s deadline in addition to the caller token; slots that never publish come back `BadTimeout`, so `CancellationToken.None` can no longer hang on a silent tag. - **Supervisory advise is retried after a thrown failure (#605 / 06-S-12)**: the writer marks an item handle advised before the call; a thrown advise (not just a non-OK status) now unmarks it, fails the write closed and re-advises on the next write. diff --git a/docs/v2/Galaxy.Performance.md b/docs/v2/Galaxy.Performance.md index 7fabe0a2c..21aec092b 100644 --- a/docs/v2/Galaxy.Performance.md +++ b/docs/v2/Galaxy.Performance.md @@ -40,6 +40,9 @@ the metrics surface instead. | `galaxy.events.dropped` | `{event}` | MxEvents discarded because the bounded channel was full (newest-dropped) | | `galaxy.events.replay_gaps` | `{gap}` | Replay-gap sentinels received — the gateway evicted events we can never see | | `galaxy.events.undecodable` | `{event}` | Data-change events carrying a value kind this driver cannot represent | +| `galaxy.events.early_parked` | `{event}` | Data-change events that overtook their SubscribeBulk reply and were parked for replay (#859) | +| `galaxy.events.early_replayed` | `{event}` | Parked early events replayed once the reply registered their handle (#859) | +| `galaxy.events.early_overflow` | `{event}` | Early events dropped because the parking bound (100 000) was reached (#859) | The invariant is `received = dispatched + dropped + replay_gaps + (in-flight in the channel)`. A replay-gap sentinel is counted as diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index d1be1bb04..7263d5c45 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -253,6 +253,9 @@ public sealed class GalaxyDriver /// Test seam: the number of driver subscriptions the registry still tracks. internal int TrackedSubscriptionCountForTests => _subscriptions.TrackedSubscriptionCount; + /// Test seam — the registry's parked early-event count (#859). + internal int ParkedEarlyEventCountForTests => _subscriptions.ParkedEventCount; + /// /// Server-pushed data-change notification. Fires from the /// 's background loop; handlers should be cheap (or queue @@ -993,6 +996,10 @@ public sealed class GalaxyDriver CancellationToken cancellationToken) { var tagCount = 0; + + // #859: the re-issued handles' initial values (the whole point of a re-snapshot) can + // overtake each SubscribeBulk reply; park them until the matching Rebind promotes them. + using var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); foreach (var (subscriptionId, oldBindings) in entries) { if (Volatile.Read(ref _runtimeGeneration) != generation) @@ -1646,6 +1653,13 @@ public sealed class GalaxyDriver // and infrastructure callers (probe watcher, deploy watcher) hit the fallback. var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds); var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs; + + // #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 + // are parked rather than dropped as stale, and close it only after TryRegisterForGeneration + // (method scope), which promotes them for in-order replay. + using var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); var results = await subscriber .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) .ConfigureAwait(false); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index 0e9433b77..2173a5a00 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -138,6 +138,10 @@ public sealed class PerPlatformProbeWatcher : IDisposable private async Task SubscribeAndRecordAsync(IReadOnlyList platforms, CancellationToken cancellationToken) { var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray(); + + // #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(); // 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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index 7eb952343..701ab73a5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -53,6 +53,15 @@ internal sealed class EventPump : IAsyncDisposable private static readonly Counter ValuesUndecodable = Meter.CreateCounter("galaxy.events.undecodable", unit: "{event}", description: "Data-change events carrying a value kind this driver cannot represent."); + private static readonly Counter EarlyEventsParked = + Meter.CreateCounter("galaxy.events.early_parked", unit: "{event}", + description: "Data-change events that arrived before their SubscribeBulk reply registered the item handle, parked for replay (#859)."); + private static readonly Counter EarlyEventsReplayed = + Meter.CreateCounter("galaxy.events.early_replayed", unit: "{event}", + description: "Parked early data-change events replayed once their item handle was registered (#859)."); + private static readonly Counter EarlyEventsOverflowed = + Meter.CreateCounter("galaxy.events.early_overflow", unit: "{event}", + description: "Early data-change events dropped because the parking bound was reached (#859)."); private readonly IGalaxySubscriber _subscriber; private readonly SubscriptionRegistry _registry; @@ -124,11 +133,28 @@ internal sealed class EventPump : IAsyncDisposable // value consistent). FullMode = BoundedChannelFullMode.Wait, SingleReader = true, - SingleWriter = true, + // Not single-writer: besides the stream read loop, a registration that promotes + // parked early events (#859) writes the WakeSentinel from the subscribing thread. + SingleWriter = false, }); _clientTag = new KeyValuePair("galaxy.client", clientName ?? ""); + _registry.ParkedEventsReady += WakeForReadyEvents; } + /// + /// Marker written into the channel to wake the dispatch loop after a registration promoted + /// parked early events (#859). Never dispatched itself; compared by reference. + /// + private static readonly MxEvent WakeSentinel = new(); + + /// + /// Wakes the dispatch loop so promoted early events replay even if no further event arrives. + /// A failed write is harmless: the channel is either full (the loop is busy and replays the + /// ready queue before its next data-change dispatch) or completed (the pump is dead, and the + /// next pump on this registry replays them). + /// + private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel); + /// /// Start consuming the event stream on a background task. Idempotent — second /// calls are no-ops while the loop is running. @@ -281,6 +307,11 @@ internal sealed class EventPump : IAsyncDisposable { await foreach (var ev in _channel.Reader.ReadAllAsync(ct).ConfigureAwait(false)) { + if (ReferenceEquals(ev, WakeSentinel)) + { + DispatchReadyEvents(); + continue; + } Dispatch(ev); EventsDispatched.Add(1, _clientTag); } @@ -313,16 +344,61 @@ internal sealed class EventPump : IAsyncDisposable // — alarms come from the gateway's session-less StreamAlarms feed // (GatewayGalaxyAlarmFeed). OnWriteComplete / OperationComplete / // OnBufferedDataChange are filtered out: write callers get their reply - // via the InvokeAsync round-trip, not via the event stream. + // via the InvokeAsync round-trip, not via the event stream. Still replay any + // promoted early events: a wake lost to a full channel relies on the next item. + DispatchReadyEvents(); return; } } private void DispatchDataChange(MxEvent ev) { - var subscribers = _registry.ResolveSubscribers(ev.ItemHandle); - if (subscribers.Count == 0) return; // stale event after unsubscribe — drop quietly + var outcome = _registry.ResolveOrPark(ev, out var subscribers); + // Replay promoted early events (#859) AFTER the resolve and BEFORE fanning this event out: + // any registration that let this event resolve has already moved its parked (older) events + // to the ready queue, so they reach the handler first and per-handle order is preserved. + DispatchReadyEvents(); + + switch (outcome) + { + case EarlyEventOutcome.Resolved: + break; + case EarlyEventOutcome.Parked: + // The handle's SubscribeBulk reply has not been registered yet — MXAccess's initial + // value can overtake it. Replayed when the registration lands. + EarlyEventsParked.Add(1, _clientTag); + return; + case EarlyEventOutcome.Overflow: + EarlyEventsOverflowed.Add(1, _clientTag); + return; + default: + return; // stale event after unsubscribe (no subscribe in flight) — drop quietly + } + + FanOut(ev, subscribers); + } + + /// + /// Dispatch every early event a registration promoted (#859). Runs only on the dispatch + /// thread, so it is serialised with live dispatch. A promoted event whose subscription was + /// removed again before the replay resolves to nobody and is dropped like any stale event. + /// + private void DispatchReadyEvents() + { + var ready = _registry.TakeReadyEvents(); + if (ready.Count == 0) return; + foreach (var early in ready) + { + var subscribers = _registry.ResolveSubscribers(early.ItemHandle); + if (subscribers.Count == 0) continue; + EarlyEventsReplayed.Add(1, _clientTag); + FanOut(early, subscribers); + } + } + + private void FanOut(MxEvent ev, IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers) + { var snapshot = ToSnapshot(ev); foreach (var (subscriptionId, fullReference) in subscribers) { @@ -377,6 +453,7 @@ internal sealed class EventPump : IAsyncDisposable { if (_disposed) return; _disposed = true; + _registry.ParkedEventsReady -= WakeForReadyEvents; _cts.Cancel(); _channel.Writer.TryComplete(); if (_loop is not null) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index 1a6ac2ff9..1bb5da9c4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Collections.Immutable; +using ZB.MOM.WW.MxGateway.Contracts.Proto; namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; @@ -79,6 +80,7 @@ internal sealed class SubscriptionRegistry _ => [subscriptionId], (_, set) => set.Add(subscriptionId)); } + PromoteParkedEvents(bindings); } /// @@ -138,6 +140,168 @@ internal sealed class SubscriptionRegistry { _bySubscriptionId.Clear(); _subscribersByItemHandle.Clear(); + lock (_parkGate) + { + _parkedByItemHandle.Clear(); + _parkedCount = 0; + _ready.Clear(); + } + } + + // ===== Early-event window (#859) ===== + // + // MXAccess delivers an initial value on Advise, and nothing orders that OnDataChange behind the + // SubscribeBulk reply the driver registers the new item handles from: + // * mxaccessgw's worker completes the STA command through a TaskCompletionSource created with + // RunContinuationsAsynchronously, so the reply is framed by a thread-pool continuation while + // the STA loop is already pumping the OnDataChange that Advise queued; the event queue is + // drained by a separate loop that frames events as soon as it is signalled + // (mxaccessgw Worker/Sta/StaCommandDispatcher.cs, Worker/Ipc/WorkerPipeSession.cs + // RunEventDrainLoopAsync). The writer's control-before-event priority only orders frames + // that are queued at the same time. + // * the reply travels back on the unary Invoke RPC and the event on the StreamEvents RPC — + // separate HTTP/2 streams with no cross-stream ordering at the client; + // * and even after the reply lands, the pump's dispatch loop runs concurrently with the code + // that turns the reply into bindings. + // So an event whose handle is not yet registered is NOT necessarily "stale after unsubscribe". + // While at least one subscribe is in flight such events are parked here (bounded) instead of + // dropped; a registration that binds a parked handle promotes its events to a ready queue that + // the EventPump's dispatch loop replays — on that same single thread, BEFORE any later event for + // the handle — so per-handle value order is preserved. When the last in-flight subscribe ends, + // whatever is still parked belonged to no subscription and is discarded, so the genuine + // stale-after-unsubscribe drop is unchanged outside the window. + + /// + /// Upper bound on events parked across all in-flight subscribes. Sized to hold the initial + /// values of a large bulk subscribe (MXAccess sends one per advised item); overflow is + /// dropped and counted by the pump, i.e. it degrades to the pre-#859 behaviour. + /// + public const int MaxParkedEvents = 100_000; + + private readonly Lock _parkGate = new(); + private readonly Dictionary> _parkedByItemHandle = new(); + private readonly Queue _ready = new(); + private int _parkedCount; + private int _pendingSubscribes; + + /// + /// Raised (outside any lock) when a registration promoted parked events to the ready queue, + /// so the can wake its dispatch loop even if no further event arrives. + /// + public event Action? ParkedEventsReady; + + /// Gets the number of subscribes currently inside their reply window. + public int PendingSubscribeCount => Volatile.Read(ref _pendingSubscribes); + + /// Gets the number of events currently parked for not-yet-registered item handles. + public int ParkedEventCount + { + get { lock (_parkGate) return _parkedCount; } + } + + /// + /// Opens the early-event window for one subscribe. Call BEFORE issuing SubscribeBulk + /// and dispose AFTER the returned handles are registered (or the attempt was abandoned): + /// while any window is open, events for unknown handles are parked rather than dropped. + /// + /// A token that closes the window when disposed (idempotent). + public IDisposable BeginPendingSubscribe() + { + Interlocked.Increment(ref _pendingSubscribes); + return new PendingSubscribeScope(this); + } + + private void EndPendingSubscribe() + { + if (Interlocked.Decrement(ref _pendingSubscribes) != 0) return; + lock (_parkGate) + { + // Re-checked under the gate: a subscribe that began after the decrement may already + // have parked its own early events, which must survive. + if (Volatile.Read(ref _pendingSubscribes) != 0) return; + _parkedByItemHandle.Clear(); + _parkedCount = 0; + } + } + + /// + /// Resolve the subscribers for 's item handle, or — when the handle is + /// unknown while a subscribe is in flight — park the event for replay. Atomic with the + /// promotion a registration performs, so an event is never parked after its handle's + /// promotion already ran (it would otherwise be stranded until the window closed). + /// + /// The data-change event. + /// The subscribers when the outcome is . + /// What happened to the event. + public EarlyEventOutcome ResolveOrPark( + MxEvent ev, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers) + { + subscribers = ResolveSubscribers(ev.ItemHandle); + if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; + if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; + + lock (_parkGate) + { + // A registration may have bound the handle between the lookup above and the gate. + subscribers = ResolveSubscribers(ev.ItemHandle); + if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; + if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; + if (_parkedCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow; + + if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var list)) + { + list = []; + _parkedByItemHandle[ev.ItemHandle] = list; + } + list.Add(ev); + _parkedCount++; + return EarlyEventOutcome.Parked; + } + } + + /// + /// Dequeue every event a registration promoted, in arrival order. Called only from the + /// EventPump's single dispatch thread. + /// + /// The promoted events; empty when none are waiting. + public IReadOnlyList TakeReadyEvents() + { + lock (_parkGate) + { + if (_ready.Count == 0) return []; + var taken = _ready.ToArray(); + _ready.Clear(); + return taken; + } + } + + private void PromoteParkedEvents(IReadOnlyList bindings) + { + var promoted = false; + lock (_parkGate) + { + if (_parkedCount == 0) return; + foreach (var binding in bindings) + { + if (binding.ItemHandle <= 0) continue; + if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var list)) continue; + foreach (var ev in list) _ready.Enqueue(ev); + _parkedCount -= list.Count; + promoted = true; + } + } + if (promoted) ParkedEventsReady?.Invoke(); + } + + private sealed class PendingSubscribeScope(SubscriptionRegistry owner) : IDisposable + { + private int _disposed; + + /// Closes this subscribe's early-event window (once). + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe(); + } } /// Snapshot every active binding for diagnostic output. @@ -189,6 +353,7 @@ internal sealed class SubscriptionRegistry _ => [subscriptionId], (_, set) => set.Add(subscriptionId)); } + PromoteParkedEvents(newBindings); } /// @@ -227,6 +392,22 @@ internal sealed class SubscriptionRegistry } } +/// Outcome of . +internal enum EarlyEventOutcome +{ + /// The handle is registered; dispatch to the returned subscribers. + Resolved, + + /// No subscribe in flight — a genuinely stale event (after unsubscribe); drop it. + Stale, + + /// Parked until the in-flight subscribe registers the handle. + Parked, + + /// Would have been parked but was reached; dropped. + Overflow, +} + /// /// One (full reference, gw item handle) pair returned by SubscribeBulk. Item handle is /// zero or negative when the gateway rejected this individual tag (bad name, duplicate); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs new file mode 100644 index 000000000..f835d7491 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs @@ -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; + +/// +/// #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. +/// +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 predicate, string because) + { + var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + while (!predicate()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException(because); + await Task.Delay(5); + } + } + + // ===== Registry ===== + + /// With no subscribe in flight an unknown handle is stale — the pre-#859 drop is kept. + [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); + } + + /// Parked events are promoted, in arrival order, when the handle is registered. + [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); + } + + /// A reconnect/re-snapshot Rebind promotes the new handles' early events too. + [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); + } + + /// Closing one of two overlapping windows keeps the other subscribe's parked events. + [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); + } + + /// Parking is bounded; past the bound the event is dropped as overflow. + [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); + } + + /// A registered handle resolves normally even while a window is open. + [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 _events = Channel.CreateUnbounded(); + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable 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); + } + + /// + /// An initial value delivered before registration reaches the handler once the handle is + /// registered — with NO further event on the stream (the wake path). + /// + [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(); + 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); + } + + /// The replayed early value precedes a later live value for the same handle. + [Fact] + public async Task Pump_replays_early_events_before_later_live_events() + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + var delivered = new ConcurrentQueue(); + 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]); + } + + /// + /// 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. + /// + [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(); + 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 ===== + + /// + /// 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. + /// + private sealed class EventBeforeReplySubscriber(Func parkedCount) : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + private int _nextHandle = 100; + + public async Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var results = new List(); + 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 itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + } + + /// The driver publishes MXAccess's initial value even when it overtook the subscribe reply. + [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(); + 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); + } +}