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

fix(galaxy): #859 review — read path window, atomic promote, session epoch, per-call windows

Browse Source 
1. ReadAsync (subscribe-once) now opens the early-event window around its
   SubscribeBulk + registration; an initial value that overtook the reply was
   dropped and the read returned BadTimeout.
2. Register/Rebind write the reverse map and promote parked events in ONE
   critical section under the park gate, and ResolveOrPark resolves under that
   gate whenever anything is held (lock-free fast path only when nothing is),
   so a later live event can no longer overtake the parked initial value. A
   resolved handle that still holds a parked list queues behind it.
3. Held events are fenced by a session epoch: each new EventPump calls
   StartNewEpoch(), discarding parked AND promoted-but-unreplayed events; an
   older pump can neither park nor replay. Item-handle reuse across a reconnect
   can no longer put an old-session value on a new tag.
4. ResubscribeAllAsync opens one window per subscription, not across the whole
   replay.
5. Promoted-but-unreplayed (ready) events count toward MaxParkedEvents.
6. Begin/End of a window run under the park gate, so N->0 discard and 0->1
   open are atomic (no inherited leftovers).
Documented that ParkedEventsReady may run under _registryGate/_syncLock and
relies on the channel's AllowSynchronousContinuations=false.

Tests use TestWait (#858); each new test was checked red against a mutation
of the fix it pins.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 10:26:28 -05:00
parent 1a31e3f584
commit f89e0cc04c
5 changed files with 505 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+82 -66
View File
@@ -996,10 +996,6 @@ 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)
@@ -1023,6 +1019,12 @@ public sealed class GalaxyDriver
.ToArray();
if (refs.Length == 0) continue;
// #859: the re-issued handles' initial values (the whole point of a re-snapshot) can
// overtake this SubscribeBulk reply; park them until the Rebind below promotes them.
// Scoped to this one iteration (the using is disposed at the end of the loop body): a
// window held across all N round-trips would park every unrelated unknown-handle event
// for the whole replay and could exhaust the bound before the genuine early values.
using var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
var results = await _subscriber!
.SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken)
.ConfigureAwait(false);
@@ -1457,75 +1459,89 @@ public sealed class GalaxyDriver
var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs;
IReadOnlyList<SubscribeResult> results;
// The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
// gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
// handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
List<TagBinding> bindings;
// #859: a read waits ONLY for the initial value, and that value can overtake the SubscribeBulk
// reply — dropped as stale, a slow-changing attribute would read BadTimeout. Open the early-event
// window before the RPC and close it right after registration (the finally below), not after the
// value wait: a window held for the whole wait would park every unrelated unknown-handle event.
var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
try
{
results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
{
// Internal deadline fired while the subscribe RPC was still in flight.
pump.OnDataChange -= handler;
var subscriber = readSubscriber;
var logger = _logger;
_ = subscribeTask.ContinueWith(t =>
// The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
// gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
// handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
try
{
if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
return late.Length == 0
? Task.CompletedTask
: subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith(
u => logger.LogWarning(u.Exception,
"GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
TaskContinuationOptions.OnlyOnFaulted);
}, TaskScheduler.Default).Unwrap();
var now = DateTime.UtcNow;
var timedOut = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < timedOut.Length; i++)
timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
return timedOut;
}
catch
{
pump.OnDataChange -= handler;
throw;
}
// Register bindings so the pump knows to dispatch events for these handles.
// Index the SubscribeBulk results once and correlate to references in O(1)
// instead of FirstOrDefault per element (O(n²) over the batch). On the
// 50k-tag soak path this turns a 2.5G-comparison loop into a single
// Dictionary build + linear scan.
var resultIndex = BuildResultIndex(results);
var bindings = new List<TagBinding>(fullReferences.Count);
for (var i = 0; i < fullReferences.Count; i++)
{
var fullRef = fullReferences[i];
var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true }
? match.ItemHandle
: 0;
bindings.Add(new TagBinding(fullRef, itemHandle));
// Tags the gw rejected up front — complete with Bad status now so the
// wait below doesn't time out on them.
if (itemHandle <= 0
&& pendingByRef.TryGetValue(fullRef, out var rejectedTcs))
results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
{
rejectedTcs.TrySetResult(new DataValueSnapshot(
Value: null,
StatusCode: StatusCodeMap.Bad,
SourceTimestampUtc: null,
ServerTimestampUtc: DateTime.UtcNow));
// Internal deadline fired while the subscribe RPC was still in flight.
pump.OnDataChange -= handler;
var subscriber = readSubscriber;
var logger = _logger;
_ = subscribeTask.ContinueWith(t =>
{
if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
return late.Length == 0
? Task.CompletedTask
: subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith(
u => logger.LogWarning(u.Exception,
"GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
TaskContinuationOptions.OnlyOnFaulted);
}, TaskScheduler.Default).Unwrap();
var now = DateTime.UtcNow;
var timedOut = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < timedOut.Length; i++)
timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
return timedOut;
}
catch
{
pump.OnDataChange -= handler;
throw;
}
// Register bindings so the pump knows to dispatch events for these handles.
// Index the SubscribeBulk results once and correlate to references in O(1)
// instead of FirstOrDefault per element (O(n²) over the batch). On the
// 50k-tag soak path this turns a 2.5G-comparison loop into a single
// Dictionary build + linear scan.
var resultIndex = BuildResultIndex(results);
bindings = new List<TagBinding>(fullReferences.Count);
for (var i = 0; i < fullReferences.Count; i++)
{
var fullRef = fullReferences[i];
var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true }
? match.ItemHandle
: 0;
bindings.Add(new TagBinding(fullRef, itemHandle));
// Tags the gw rejected up front — complete with Bad status now so the
// wait below doesn't time out on them.
if (itemHandle <= 0
&& pendingByRef.TryGetValue(fullRef, out var rejectedTcs))
{
rejectedTcs.TrySetResult(new DataValueSnapshot(
Value: null,
StatusCode: StatusCodeMap.Bad,
SourceTimestampUtc: null,
ServerTimestampUtc: DateTime.UtcNow));
}
}
if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
{
pump.OnDataChange -= handler;
await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
}
}
if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
finally
{
pump.OnDataChange -= handler;
await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
earlyEventWindow.Dispose();
}
try
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
+19 -7
View File
@@ -138,9 +138,16 @@ internal sealed class EventPump : IAsyncDisposable
SingleWriter = false,
});
_clientTag = new KeyValuePair<string, object?>("galaxy.client", clientName ?? "<unknown>");
// One pump per gateway session: a new epoch discards whatever an earlier session left held
// (#859 review — item handles are reused across sessions), and this pump never parks or
// replays under any other epoch.
_epoch = _registry.StartNewEpoch();
_registry.ParkedEventsReady += WakeForReadyEvents;
}
private readonly int _epoch;
/// <summary>
/// Marker written into the channel to wake the dispatch loop after a registration promoted
/// parked early events (#859). Never dispatched itself; compared by reference.
@@ -149,9 +156,12 @@ internal sealed class EventPump : IAsyncDisposable
/// <summary>
/// Wakes the dispatch loop so promoted early events replay even if no further event arrives.
/// Runs on the registering thread, possibly under the driver's <c>_registryGate</c> or the probe
/// watcher's <c>_syncLock</c> — safe because the channel keeps the default
/// <c>AllowSynchronousContinuations = false</c>, so the write never runs the dispatch loop inline.
/// 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).
/// ready queue before its next dispatch) or completed (the pump is dead; its held events
/// belong to a session the next pump's new epoch discards).
/// </summary>
private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel);
@@ -353,11 +363,13 @@ internal sealed class EventPump : IAsyncDisposable
private void DispatchDataChange(MxEvent ev)
{
var outcome = _registry.ResolveOrPark(ev, out var subscribers);
var outcome = _registry.ResolveOrPark(ev, _epoch, 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.
// Replay promoted early events (#859) AFTER the resolve and BEFORE fanning this event out.
// A registration writes the reverse map and promotes its parked events in ONE critical
// section under the registry's park gate, and ResolveOrPark resolves under that gate
// whenever anything is held. So if this event resolved, every older parked event for its
// handle is already in the ready queue, and replaying it first preserves per-handle order.
DispatchReadyEvents();
switch (outcome)
@@ -386,7 +398,7 @@ internal sealed class EventPump : IAsyncDisposable
/// </summary>
private void DispatchReadyEvents()
{
var ready = _registry.TakeReadyEvents();
var ready = _registry.TakeReadyEvents(_epoch);
if (ready.Count == 0) return;
foreach (var early in ready)
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
+165 -73
View File
@@ -70,17 +70,27 @@ internal sealed class SubscriptionRegistry
/// <param name="bindings">The tag bindings for the subscription.</param>
public void Register(long subscriptionId, IReadOnlyList<TagBinding> bindings)
{
var entry = new SubscriptionEntry(subscriptionId, bindings);
_bySubscriptionId[subscriptionId] = entry;
foreach (var binding in bindings)
bool promoted;
// The reverse-map writes and the early-event promotion are ONE critical section under the
// park gate (#859 review): a dispatch that resolves a handle through the map must never
// run between the write and the promotion, or a later live event would overtake the
// parked initial value. ResolveOrPark takes the same gate whenever anything is held.
lock (_parkGate)
{
if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
_subscribersByItemHandle.AddOrUpdate(
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
var entry = new SubscriptionEntry(subscriptionId, bindings);
_bySubscriptionId[subscriptionId] = entry;
foreach (var binding in bindings)
{
if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
_subscribersByItemHandle.AddOrUpdate(
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
}
AfterMapWriteForTests?.Invoke();
promoted = PromoteParkedEventsLocked(bindings);
}
PromoteParkedEvents(bindings);
if (promoted) ParkedEventsReady?.Invoke();
}
/// <summary>
@@ -143,8 +153,8 @@ internal sealed class SubscriptionRegistry
lock (_parkGate)
{
_parkedByItemHandle.Clear();
_parkedCount = 0;
_ready.Clear();
Volatile.Write(ref _heldCount, 0);
}
}
@@ -170,60 +180,110 @@ internal sealed class SubscriptionRegistry
// 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.
//
// Item handles are only meaningful within one gateway session, and a new session reuses small
// integers. Every held event therefore belongs to an EPOCH: each EventPump (one per session —
// reconnect replay recreates it) starts a new epoch, which discards everything held, and an
// event parked or replayed by a pump of an older epoch is dropped. Otherwise an old-session
// value could be promoted onto whichever tag now owns the reused handle number.
/// <summary>
/// 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.
/// Upper bound on events held (parked + promoted-but-not-yet-replayed) 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.
/// </summary>
public const int MaxParkedEvents = 100_000;
private readonly Lock _parkGate = new();
private readonly Dictionary<int, List<MxEvent>> _parkedByItemHandle = new();
private readonly Queue<MxEvent> _ready = new();
private int _parkedCount;
// Parked + ready. Written only under _parkGate; read lock-free (Volatile) by the dispatch fast path.
private int _heldCount;
// Open windows. Written only under _parkGate (so the 0→N / N→0 transitions and the discard of
// the previous window's leftovers are atomic); read lock-free by the dispatch fast path.
private int _pendingSubscribes;
private int _epoch;
/// <summary>
/// Raised (outside any lock) when a registration promoted parked events to the ready queue,
/// so the <see cref="EventPump"/> can wake its dispatch loop even if no further event arrives.
/// Test seam (#859 review): invoked inside the registration critical section, between the
/// reverse-map writes and the promotion of parked events.
/// </summary>
internal Action? AfterMapWriteForTests { get; set; }
/// <summary>
/// Raised when a registration promoted parked events to the ready queue, so the
/// <see cref="EventPump"/> can wake its dispatch loop even if no further event arrives.
/// Raised after the park gate is released but still on the registering thread, which may hold
/// the driver's <c>_registryGate</c> or the probe watcher's <c>_syncLock</c>: a handler must
/// only signal (the pump's handler is a channel <c>TryWrite</c>, safe there because the
/// channel keeps the default <c>AllowSynchronousContinuations = false</c>, so the dispatch
/// loop never runs inline under those locks).
/// </summary>
public event Action? ParkedEventsReady;
/// <summary>Gets the number of subscribes currently inside their reply window.</summary>
public int PendingSubscribeCount => Volatile.Read(ref _pendingSubscribes);
/// <summary>Gets the number of events currently parked for not-yet-registered item handles.</summary>
public int ParkedEventCount
/// <summary>Gets the number of events currently held (parked or awaiting replay).</summary>
public int ParkedEventCount => Volatile.Read(ref _heldCount);
/// <summary>
/// Starts a new session epoch: discards every held event and makes events from any earlier
/// epoch unparkable and unreplayable. Called by each new <see cref="EventPump"/>.
/// </summary>
/// <returns>The new epoch.</returns>
public int StartNewEpoch()
{
get { lock (_parkGate) return _parkedCount; }
lock (_parkGate)
{
_parkedByItemHandle.Clear();
_ready.Clear();
Volatile.Write(ref _heldCount, 0);
return ++_epoch;
}
}
/// <summary>
/// Opens the early-event window for one subscribe. Call BEFORE issuing <c>SubscribeBulk</c>
/// 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.
/// and dispose AFTER the returned handles are registered (or the attempt was abandoned) —
/// not later: while any window is open, events for unknown handles are parked, not dropped.
/// </summary>
/// <returns>A token that closes the window when disposed (idempotent).</returns>
public IDisposable BeginPendingSubscribe()
{
Interlocked.Increment(ref _pendingSubscribes);
lock (_parkGate)
{
// 0→1: anything still parked belongs to an earlier, closed window. EndPendingSubscribe
// already discards on N→0 under this same gate, so this is belt-and-braces.
if (_pendingSubscribes == 0) DiscardParkedLocked();
Volatile.Write(ref _pendingSubscribes, _pendingSubscribes + 1);
}
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;
Volatile.Write(ref _pendingSubscribes, _pendingSubscribes - 1);
// N→0: what is still parked was claimed by no registration. Promoted (ready) events are
// kept — they belong to a registered handle and are awaiting replay.
if (_pendingSubscribes == 0) DiscardParkedLocked();
}
}
private void DiscardParkedLocked()
{
if (_parkedByItemHandle.Count == 0) return;
var parked = 0;
foreach (var list in _parkedByItemHandle.Values) parked += list.Count;
_parkedByItemHandle.Clear();
Volatile.Write(ref _heldCount, _heldCount - parked);
}
/// <summary>
/// Resolve the subscribers for <paramref name="ev"/>'s item handle, or — when the handle is
/// unknown while a subscribe is in flight — park the event for replay. Atomic with the
@@ -231,22 +291,49 @@ internal sealed class SubscriptionRegistry
/// promotion already ran (it would otherwise be stranded until the window closed).
/// </summary>
/// <param name="ev">The data-change event.</param>
/// <param name="epoch">The calling pump's epoch (<see cref="StartNewEpoch"/>).</param>
/// <param name="subscribers">The subscribers when the outcome is <see cref="EarlyEventOutcome.Resolved"/>.</param>
/// <returns>What happened to the event.</returns>
public EarlyEventOutcome ResolveOrPark(
MxEvent ev, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers)
MxEvent ev, int epoch, 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)
// Lock-free fast path ONLY while nothing is held: then no older event for this handle can be
// waiting, so resolving through the map cannot overtake one. While anything is held, the
// resolve happens under the gate, i.e. never inside a registration's map-write→promotion
// critical section.
if (Volatile.Read(ref _heldCount) == 0)
{
// 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;
}
lock (_parkGate)
{
if (epoch != _epoch)
{
subscribers = [];
return EarlyEventOutcome.Stale; // a dying pump of an earlier session
}
subscribers = ResolveSubscribers(ev.ItemHandle);
if (subscribers.Count > 0)
{
// Defensive: registration promotes atomically with its map writes, so a resolved
// handle should hold no parked list — but if one ever does, queue behind it rather
// than overtake it (promotion moves the whole list in order). Not bounded: dropping
// this newer value would leave the older parked one as the node's last word.
if (_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var pending))
{
pending.Add(ev);
Volatile.Write(ref _heldCount, _heldCount + 1);
subscribers = [];
return EarlyEventOutcome.Parked;
}
return EarlyEventOutcome.Resolved;
}
if (_pendingSubscribes == 0) return EarlyEventOutcome.Stale;
if (_heldCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow;
if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var list))
{
@@ -254,43 +341,42 @@ internal sealed class SubscriptionRegistry
_parkedByItemHandle[ev.ItemHandle] = list;
}
list.Add(ev);
_parkedCount++;
Volatile.Write(ref _heldCount, _heldCount + 1);
return EarlyEventOutcome.Parked;
}
}
/// <summary>
/// Dequeue every event a registration promoted, in arrival order. Called only from the
/// EventPump's single dispatch thread.
/// EventPump's single dispatch thread. A pump of an earlier epoch gets nothing.
/// </summary>
/// <param name="epoch">The calling pump's epoch.</param>
/// <returns>The promoted events; empty when none are waiting.</returns>
public IReadOnlyList<MxEvent> TakeReadyEvents()
public IReadOnlyList<MxEvent> TakeReadyEvents(int epoch)
{
if (Volatile.Read(ref _heldCount) == 0) return [];
lock (_parkGate)
{
if (_ready.Count == 0) return [];
if (epoch != _epoch || _ready.Count == 0) return [];
var taken = _ready.ToArray();
_ready.Clear();
Volatile.Write(ref _heldCount, _heldCount - taken.Length);
return taken;
}
}
private void PromoteParkedEvents(IReadOnlyList<TagBinding> bindings)
private bool PromoteParkedEventsLocked(IReadOnlyList<TagBinding> bindings)
{
if (_parkedByItemHandle.Count == 0) return false;
var promoted = false;
lock (_parkGate)
foreach (var binding in bindings)
{
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 (binding.ItemHandle <= 0) continue;
if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var list)) continue;
foreach (var ev in list) _ready.Enqueue(ev); // stays counted in _heldCount until replayed
promoted = true;
}
if (promoted) ParkedEventsReady?.Invoke();
return promoted;
}
private sealed class PendingSubscribeScope(SubscriptionRegistry owner) : IDisposable
@@ -328,32 +414,38 @@ internal sealed class SubscriptionRegistry
/// <param name="newBindings">The new tag bindings after reconnection.</param>
public void Rebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings)
{
if (!_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) return;
// Drop this subscription from every reverse-map set it currently appears in. The
// pre-reconnect item handles are stale once the gw re-issues fresh ones.
foreach (var binding in oldEntry.Bindings)
bool promoted;
lock (_parkGate) // map writes + promotion atomic — see Register
{
if (binding.ItemHandle <= 0) continue;
if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
var remaining = set.Remove(subscriptionId);
if (remaining.IsEmpty)
if (!_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) return;
// Drop this subscription from every reverse-map set it currently appears in. The
// pre-reconnect item handles are stale once the gw re-issues fresh ones.
foreach (var binding in oldEntry.Bindings)
{
_subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
if (binding.ItemHandle <= 0) continue;
if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
var remaining = set.Remove(subscriptionId);
if (remaining.IsEmpty)
{
_subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
}
else _subscribersByItemHandle[binding.ItemHandle] = remaining;
}
else _subscribersByItemHandle[binding.ItemHandle] = remaining;
}
_bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, newBindings);
foreach (var binding in newBindings)
{
if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
_subscribersByItemHandle.AddOrUpdate(
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
_bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, newBindings);
foreach (var binding in newBindings)
{
if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
_subscribersByItemHandle.AddOrUpdate(
binding.ItemHandle,
_ => [subscriptionId],
(_, set) => set.Add(subscriptionId));
}
AfterMapWriteForTests?.Invoke();
promoted = PromoteParkedEventsLocked(newBindings);
}
PromoteParkedEvents(newBindings);
if (promoted) ParkedEventsReady?.Invoke();
}
/// <summary>