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

fix(galaxy): #859 second review + #861 — window-sequence trim, per-session handles, straddling subscribes

Browse Source 
I-2: unclaimed parked events are trimmed by window sequence. Each window gets a
monotonically increasing sequence at Begin, each parked event is stamped with
the newest one, and on every End events stamped before the oldest open
window's sequence are dropped (amortised O(1) via a park-order queue). Under
concurrent reads the open count rarely hit 0, so trailing events used to pile
up to the bound and push genuine early values into overflow.

I-1 / #861: SubscriptionRegistry.StartNewSession() — called FIRST in
ReplayAsync on every reconnect — clears held events, invalidates the reverse
map (subscriptions kept) and bumps a session number. New-session events whose
number an old binding still held now park until the replay's per-subscription
Rebind promotes them, instead of resolving to the wrong tag. TryRegister /
TryRebind refuse a stale session atomically under the park gate: a
SubscribeAsync whose reply straddled a reconnect is re-issued on the new
session (max 3 attempts), a subscribe-once read returns BadCommunicationError,
ResubscribeAllAsync stops if the session changes again. Old-session handle
numbers are never unsubscribed (Remove reports whether the entry belongs to
the current session; the read's late-reply path checks too), and an old
session's pump can no longer resolve through the map (fast path checks the
session).

I-3 (interim): a completed UnsubscribeBulk purges events parked for its
handles (driver + probe watcher). Full fix needs a worker sequence on the
SubscribeBulk reply (mxaccessgw, optional) — documented.

M-1: a resolved handle with a stranded parked list promotes it and resolves
the live event (never parks it). M-2: ResolveOrPark returns the ready events
to replay first, one gate acquisition per event while anything is held.
M-3: SubscribeAsync/read dispose the window before the orphan-release RPC.
M-5: ResolveOrPark remarks document the single-dispatching-pump invariant.

Tests (TestWait): overlapping windows don't accumulate, purge, stranded list,
session fence + map invalidation, stale-session registration refused,
end-to-end replay with swapped handle numbers and event-before-reply,
subscribe and read straddling a reconnect, read deadline closes its window.
Each was checked red against a mutation of the fix it pins.

Refs #859, #861.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 10:41:16 -05:00
parent f89e0cc04c
commit d08e900203
6 changed files with 789 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+153 -35
View File
@@ -256,6 +256,9 @@ public sealed class GalaxyDriver
/// <summary>Test seam — the registry's parked early-event count (#859).</summary>
internal int ParkedEarlyEventCountForTests => _subscriptions.ParkedEventCount;
/// <summary>Test seam — the driver's subscription registry (#859/#861 window + session tests).</summary>
internal SubscriptionRegistry SubscriptionsForTests => _subscriptions;
/// <summary>
/// Server-pushed data-change notification. Fires from the
/// <see cref="EventPump"/>'s background loop; handlers should be cheap (or queue
@@ -644,13 +647,37 @@ public sealed class GalaxyDriver
/// Register <paramref name="bindings"/> only while <paramref name="generation"/> is still the
/// live runtime. Atomic with the teardown's registry Clear (both under <see cref="_registryGate"/>).
/// </summary>
private bool TryRegisterForGeneration(long subscriptionId, IReadOnlyList<TagBinding> bindings, int generation)
private RegisterOutcome TryRegisterForGeneration(
long subscriptionId, IReadOnlyList<TagBinding> bindings, int generation, int session)
{
lock (_registryGate)
{
if (Volatile.Read(ref _runtimeGeneration) != generation) return false;
_subscriptions.Register(subscriptionId, bindings);
return true;
if (Volatile.Read(ref _runtimeGeneration) != generation) return RegisterOutcome.TornDown;
// #861: the reply may have straddled a reconnect — its handles then belong to the dead
// session and collide with the new session's numbers. Checked atomically with
// StartNewSession inside the registry.
return _subscriptions.TryRegister(subscriptionId, bindings, session)
? RegisterOutcome.Registered
: RegisterOutcome.SessionChanged;
}
}
private enum RegisterOutcome { Registered, TornDown, SessionChanged }
/// <summary>
/// UnsubscribeBulk for <paramref name="itemHandles"/>, then purge any event parked for them
/// (#859 I-3 interim — a trailing event must not be promoted onto a reused handle number).
/// </summary>
private async Task UnsubscribeAndPurgeAsync(
IGalaxySubscriber subscriber, IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
{
try
{
await subscriber.UnsubscribeBulkAsync(itemHandles, cancellationToken).ConfigureAwait(false);
}
finally
{
_subscriptions.PurgeParked(itemHandles);
}
}
@@ -667,7 +694,7 @@ public sealed class GalaxyDriver
{
var live = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
if (live.Length == 0) return;
try { await subscriber.UnsubscribeBulkAsync(live, CancellationToken.None).ConfigureAwait(false); }
try { await UnsubscribeAndPurgeAsync(subscriber, live, CancellationToken.None).ConfigureAwait(false); }
catch (Exception ex)
{
_logger.LogWarning(ex,
@@ -945,6 +972,15 @@ public sealed class GalaxyDriver
// anyway would start a fresh pump underneath the shutdown.
cancellationToken.ThrowIfCancellationRequested();
if (_subscriber is null) return;
// The session was recreated: every item handle the registry holds is dead and its number
// may be reused by the new session (#859 I-1 / #861). Invalidate the reverse map (the
// subscriptions stay; the per-iteration Rebind below re-binds each one, and new-session
// events park meanwhile instead of resolving to the old owner of a reused number), discard
// held events, and fence any subscribe whose reply straddled the reconnect. Done FIRST and
// unconditionally — before the pump restart and even with nothing to replay.
_subscriptions.StartNewSession();
var entries = _subscriptions.SnapshotEntries();
if (entries.Count == 0) return;
@@ -1025,6 +1061,7 @@ public sealed class GalaxyDriver
// 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 session = _subscriptions.CurrentSession;
var results = await _subscriber!
.SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken)
.ConfigureAwait(false);
@@ -1045,7 +1082,15 @@ public sealed class GalaxyDriver
lock (_registryGate)
{
if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount;
_subscriptions.Rebind(subscriptionId, newBindings);
// #861: another reconnect during this round-trip makes these handles dead too; the
// newer session's own replay re-subscribes everything, so stop here.
if (!_subscriptions.TryRebind(subscriptionId, newBindings, session))
{
_logger.LogDebug(
"GalaxyDriver {InstanceId} re-subscribe abandoned — the gateway session changed again.",
_driverInstanceId);
return tagCount;
}
}
tagCount += refs.Length;
}
@@ -1466,6 +1511,7 @@ public sealed class GalaxyDriver
// 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();
var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
try
{
// The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
@@ -1485,10 +1531,13 @@ public sealed class GalaxyDriver
_ = subscribeTask.ContinueWith(t =>
{
if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
// A late reply from a session that has since been replaced carries dead handle
// numbers the new session may reuse — never unsubscribe those (#861).
if (_subscriptions.CurrentSession != session) 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(
: UnsubscribeAndPurgeAsync(subscriber, late, CancellationToken.None).ContinueWith(
u => logger.LogWarning(u.Exception,
"GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
TaskContinuationOptions.OnlyOnFaulted);
@@ -1532,11 +1581,28 @@ public sealed class GalaxyDriver
ServerTimestampUtc: DateTime.UtcNow));
}
}
if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
var registered = TryRegisterForGeneration(subscriptionId, bindings, generation, session);
if (registered != RegisterOutcome.Registered)
{
pump.OnDataChange -= handler;
await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
earlyEventWindow.Dispose(); // before any further round-trip
if (registered == RegisterOutcome.TornDown)
{
await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
}
// #861: the reply straddled a reconnect. Its handles died with the old session and
// their numbers may already belong to new-session items, so they are neither
// registered nor unsubscribed. A read is one-shot: report the lost session per tag.
_logger.LogInformation(
"GalaxyDriver {InstanceId} subscribe-once read of {Count} tag(s) straddled a gateway reconnect — returning BadCommunicationError.",
_driverInstanceId, fullReferences.Count);
var lostAt = DateTime.UtcNow;
var lost = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < lost.Length; i++)
lost[i] = new DataValueSnapshot(null, StatusCodeMap.BadCommunicationError, null, lostAt);
return lost;
}
}
finally
@@ -1575,13 +1641,16 @@ public sealed class GalaxyDriver
// Drop the bindings + unsubscribe the live handles. UnsubscribeBulkAsync's
// failure isn't fatal — the registry is already cleared, so any straggling
// event from the gw would be a no-op fan-out.
_subscriptions.Remove(subscriptionId);
var removed = _subscriptions.Remove(subscriptionId, out var liveInCurrentSession);
// A reconnect during the wait left these handles in the dead session; their numbers may now
// be new-session items, so they must not be unsubscribed (#861).
var staleSession = removed is not null && !liveInCurrentSession;
var liveHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
if (liveHandles.Length > 0)
if (liveHandles.Length > 0 && !staleSession)
{
try
{
await readSubscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None)
await UnsubscribeAndPurgeAsync(readSubscriber, liveHandles, CancellationToken.None)
.ConfigureAwait(false);
}
catch (Exception ex)
@@ -1653,7 +1722,14 @@ public sealed class GalaxyDriver
{
// Empty subscriptions register but never bind anything — keeps Unsubscribe
// symmetric for callers that conditionally add tags later.
if (!TryRegisterForGeneration(subscriptionId, [], generation)) throw TornDownDuringSubscribe();
RegisterOutcome outcome;
do
{
// No handles, so a session change only means "read the session again".
outcome = TryRegisterForGeneration(subscriptionId, [], generation, _subscriptions.CurrentSession);
}
while (outcome == RegisterOutcome.SessionChanged);
if (outcome == RegisterOutcome.TornDown) throw TornDownDuringSubscribe();
return new GalaxySubscriptionHandle(subscriptionId);
}
@@ -1670,16 +1746,63 @@ public sealed class GalaxyDriver
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);
for (var attempt = 1; ; attempt++)
{
// #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; it closes once the registration (which promotes
// them for in-order replay) is done.
var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
try
{
var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
var results = await subscriber
.SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken)
.ConfigureAwait(false);
var bindings = BuildSubscribeBindings(dialledRefs, results);
switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session))
{
case RegisterOutcome.Registered:
_ = pump; // keep the pump alive for the subscription's lifetime
return new GalaxySubscriptionHandle(subscriptionId);
case RegisterOutcome.TornDown:
// The teardown cleared the registry while SubscribeBulk was in flight; registering
// now would carry dead handles into the next runtime. Release what we got and fail.
earlyEventWindow.Dispose(); // M-3: not held across the release round-trip
await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
default:
// #861: the reply straddled a gateway reconnect, so its handles died with the old
// session (and are NOT unsubscribed — the new session may reuse the numbers). The
// reconnect replay only re-subscribes what was registered when it snapshotted, so
// this subscription would be lost: issue it again on the new session.
if (attempt >= MaxSubscribeSessionAttempts)
{
throw new InvalidOperationException(
$"Galaxy subscribe straddled {attempt} consecutive gateway reconnects — giving up; retry later.");
}
_logger.LogInformation(
"GalaxyDriver {InstanceId} subscribe of {Count} tag(s) straddled a gateway reconnect — re-issuing on the new session (attempt {Attempt}).",
_driverInstanceId, dialledRefs.Length, attempt + 1);
continue;
}
}
finally
{
earlyEventWindow.Dispose();
}
}
}
/// <summary>Bound on re-issuing a subscribe whose reply straddled a gateway reconnect (#861).</summary>
private const int MaxSubscribeSessionAttempts = 3;
private List<TagBinding> BuildSubscribeBindings(string[] dialledRefs, IReadOnlyList<SubscribeResult> results)
{
// Build the binding list in input order. Failed entries (gw rejected the tag) are
// recorded with a non-positive ItemHandle so the caller can detect partial failure
// by inspecting the returned handle's diagnostic context — full per-tag error
@@ -1703,16 +1826,7 @@ public sealed class GalaxyDriver
fullRef, match?.ErrorMessage ?? "<no result returned>");
}
}
if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
{
// The teardown cleared the registry while SubscribeBulk was in flight; registering now
// would carry dead handles into the next runtime. Release what we got and fail the call.
await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false);
throw TornDownDuringSubscribe();
}
_ = pump; // keep the pump alive for the subscription's lifetime
return new GalaxySubscriptionHandle(subscriptionId);
return bindings;
}
/// <inheritdoc />
@@ -1727,15 +1841,19 @@ public sealed class GalaxyDriver
nameof(handle));
}
var bindings = _subscriptions.Remove(gsh.SubscriptionId);
var bindings = _subscriptions.Remove(gsh.SubscriptionId, out var liveInCurrentSession);
if (bindings is null) return; // already removed or never registered
// Not yet rebound after a reconnect: its handles died with the old session and the new one may
// have reused the numbers for other items — unsubscribing them would unadvise those (#861).
if (!liveInCurrentSession) return;
var liveItemHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
if (liveItemHandles.Length == 0 || _subscriber is null) return;
try
{
await _subscriber.UnsubscribeBulkAsync(liveItemHandles, cancellationToken).ConfigureAwait(false);
await UnsubscribeAndPurgeAsync(_subscriber, liveItemHandles, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs
+6
View File
@@ -260,6 +260,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable
"PerPlatformProbeWatcher unsubscribe failed for {Count} probe(s); aggregator entries already cleared.",
toRemove.Count);
}
finally
{
// #859 I-3 interim: no trailing event for a removed probe handle may be promoted onto
// a reused handle number.
_registry?.PurgeParked([.. toRemove.Select(t => t.ItemHandle)]);
}
}
if (toAdd.Count == 0) return;
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
+16 -15
View File
@@ -139,14 +139,14 @@ internal sealed class EventPump : IAsyncDisposable
});
_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();
// One pump per gateway session. The owner calls SubscriptionRegistry.StartNewSession() BEFORE
// creating the pump for a new session (reconnect); this pump binds to that session and never
// parks or replays under any other (#859 review — item handles are reused across sessions).
_session = _registry.CurrentSession;
_registry.ParkedEventsReady += WakeForReadyEvents;
}
private readonly int _epoch;
private readonly int _session;
/// <summary>
/// Marker written into the channel to wake the dispatch loop after a registration promoted
@@ -161,7 +161,7 @@ internal sealed class EventPump : IAsyncDisposable
/// <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 dispatch) or completed (the pump is dead; its held events
/// belong to a session the next pump's new epoch discards).
/// belong to a session the next StartNewSession discards).
/// </summary>
private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel);
@@ -363,14 +363,14 @@ internal sealed class EventPump : IAsyncDisposable
private void DispatchDataChange(MxEvent ev)
{
var outcome = _registry.ResolveOrPark(ev, _epoch, out var subscribers);
var outcome = _registry.ResolveOrPark(ev, _session, out var subscribers, out var replayFirst);
// 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();
// Replay promoted early events (#859) 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 — and takes the ready queue — under that
// same gate whenever anything is held. So if this event resolved, every older parked event
// for its handle is in replayFirst, and replaying it first preserves per-handle order.
ReplayEarly(replayFirst);
switch (outcome)
{
@@ -396,9 +396,10 @@ internal sealed class EventPump : IAsyncDisposable
/// 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.
/// </summary>
private void DispatchReadyEvents()
private void DispatchReadyEvents() => ReplayEarly(_registry.TakeReadyEvents(_session));
private void ReplayEarly(IReadOnlyList<MxEvent> ready)
{
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
+290 -141
View File
@@ -68,17 +68,70 @@ internal sealed class SubscriptionRegistry
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="bindings">The tag bindings for the subscription.</param>
public void Register(long subscriptionId, IReadOnlyList<TagBinding> bindings)
public void Register(long subscriptionId, IReadOnlyList<TagBinding> bindings) =>
Bind(subscriptionId, bindings, requireExisting: false, expectedSession: null);
/// <summary>
/// <see cref="Register"/>, but only if the gateway session is still
/// <paramref name="expectedSession"/> — the session the SubscribeBulk that produced
/// <paramref name="bindings"/> was issued on (#861). Item handles are per session and the next
/// session reuses the numbers, so registering a reply that straddled a reconnect would bind
/// dead handles that collide with live ones.
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="bindings">The tag bindings for the subscription.</param>
/// <param name="expectedSession">The <see cref="CurrentSession"/> captured before the SubscribeBulk.</param>
/// <returns>False (nothing registered) when the session changed in the meantime.</returns>
public bool TryRegister(long subscriptionId, IReadOnlyList<TagBinding> bindings, int expectedSession) =>
Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindResult.Bound;
/// <summary>
/// Replace an existing subscription's bindings with the item handles a post-reconnect
/// SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new
/// handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the
/// subscription id is unknown (it was unsubscribed during the reconnect window).
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="newBindings">The new tag bindings after reconnection.</param>
public void Rebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings) =>
Bind(subscriptionId, newBindings, requireExisting: true, expectedSession: null);
/// <summary><see cref="Rebind"/>, but only while the session is still <paramref name="expectedSession"/> (#861).</summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="newBindings">The new tag bindings.</param>
/// <param name="expectedSession">The session the SubscribeBulk was issued on.</param>
/// <returns>False when the session changed (nothing rebound); true otherwise, including the unknown-id no-op.</returns>
public bool TryRebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings, int expectedSession) =>
Bind(subscriptionId, newBindings, requireExisting: true, expectedSession) != BindResult.SessionChanged;
private enum BindResult { Bound, UnknownId, SessionChanged }
private BindResult Bind(
long subscriptionId, IReadOnlyList<TagBinding> bindings, bool requireExisting, int? expectedSession)
{
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.
// parked initial value. ResolveOrPark takes the same gate whenever anything is held. The
// session check is in here too, so it is atomic with StartNewSession's invalidation.
lock (_parkGate)
{
var entry = new SubscriptionEntry(subscriptionId, bindings);
_bySubscriptionId[subscriptionId] = entry;
if (expectedSession is { } session && session != _session) return BindResult.SessionChanged;
if (_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry))
{
// Drop this subscription from every reverse-map set it currently appears in. Only
// this id is removed, so another subscription that now owns the same handle number
// (a reused handle in a new session) keeps its entry.
RemoveFromReverseMap(oldEntry);
}
else if (requireExisting)
{
return BindResult.UnknownId;
}
_bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, bindings, _session);
foreach (var binding in bindings)
{
if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
@@ -91,6 +144,22 @@ internal sealed class SubscriptionRegistry
promoted = PromoteParkedEventsLocked(bindings);
}
if (promoted) ParkedEventsReady?.Invoke();
return BindResult.Bound;
}
private void RemoveFromReverseMap(SubscriptionEntry entry)
{
foreach (var binding in entry.Bindings)
{
if (binding.ItemHandle <= 0) continue;
if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
var remaining = set.Remove(entry.SubscriptionId);
if (remaining.IsEmpty)
{
_subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
}
else _subscribersByItemHandle[binding.ItemHandle] = remaining;
}
}
/// <summary>
@@ -99,23 +168,27 @@ internal sealed class SubscriptionRegistry
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <returns>The bindings for the subscription, or null if not found.</returns>
public IReadOnlyList<TagBinding>? Remove(long subscriptionId)
public IReadOnlyList<TagBinding>? Remove(long subscriptionId) => Remove(subscriptionId, out _);
/// <summary>
/// Remove a subscription, also reporting whether its handles belong to the CURRENT gateway
/// session. A subscription not yet rebound after a reconnect still carries the previous
/// session's handles, whose numbers the new session reuses — the caller must not send those
/// to <c>UnsubscribeBulk</c>, or it would unadvise some other live item.
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="liveInCurrentSession">True when the bindings' handles were issued by the current session.</param>
/// <returns>The bindings for the subscription, or null if not found.</returns>
public IReadOnlyList<TagBinding>? Remove(long subscriptionId, out bool liveInCurrentSession)
{
if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null;
foreach (var binding in entry.Bindings)
lock (_parkGate)
{
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;
liveInCurrentSession = false;
if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null;
RemoveFromReverseMap(entry);
liveInCurrentSession = entry.Session == _session;
return entry.Bindings;
}
return entry.Bindings;
}
/// <summary>
@@ -140,7 +213,6 @@ internal sealed class SubscriptionRegistry
return result;
}
/// <summary>
/// Drop every subscription and the whole reverse map. Used by the driver's runtime teardown:
/// once the gateway session is closed every item handle in here is dead, and the host
@@ -148,13 +220,11 @@ internal sealed class SubscriptionRegistry
/// </summary>
public void Clear()
{
_bySubscriptionId.Clear();
_subscribersByItemHandle.Clear();
lock (_parkGate)
{
_parkedByItemHandle.Clear();
_ready.Clear();
Volatile.Write(ref _heldCount, 0);
_bySubscriptionId.Clear();
_subscribersByItemHandle.Clear();
DiscardAllHeldLocked();
}
}
@@ -177,15 +247,21 @@ internal sealed class SubscriptionRegistry
// 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.
// the handle — so per-handle value order is preserved.
//
// 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.
// Unclaimed parked events are trimmed by WINDOW SEQUENCE, not only when every window has
// closed (#859 second review — concurrent reads keep some window open almost permanently):
// each window gets a monotonically increasing sequence when it opens, and each parked event is
// stamped with the newest sequence at park time. An event belonging to window W is generated
// after W opened, so its stamp is >= W's sequence; once every window with a sequence <= the
// stamp has closed, nothing can claim it and it is discarded.
//
// Item handles are only meaningful within one gateway SESSION, and a new session reuses small
// integers. StartNewSession (called on reconnect, before the replay re-subscribes) discards
// every held event, invalidates the reverse map (so a new-session event can never resolve to
// the tag that owned that number in the old session — it parks until the replay's Rebind),
// and fences: an event parked or replayed by an older session's pump is dropped, and a
// subscribe whose reply straddled the reconnect cannot register its dead handles (#861).
/// <summary>
/// Upper bound on events held (parked + promoted-but-not-yet-replayed) across all in-flight
@@ -196,16 +272,18 @@ internal sealed class SubscriptionRegistry
public const int MaxParkedEvents = 100_000;
private readonly Lock _parkGate = new();
private readonly Dictionary<int, List<MxEvent>> _parkedByItemHandle = new();
private readonly Dictionary<int, Queue<ParkedEvent>> _parkedByItemHandle = new();
private readonly Queue<ParkedEvent> _parkOrder = new(); // every parked event, in park order (trim index)
private readonly Queue<MxEvent> _ready = new();
private readonly SortedSet<long> _openWindows = new();
private long _windowSequence;
// 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.
// Open windows. Written only under _parkGate; read lock-free by the dispatch fast path.
private int _pendingSubscribes;
private int _epoch;
private int _session;
/// <summary>
/// Test seam (#859 review): invoked inside the registration critical section, between the
@@ -213,6 +291,27 @@ internal sealed class SubscriptionRegistry
/// </summary>
internal Action? AfterMapWriteForTests { get; set; }
/// <summary>
/// Test seam (M-1): parks <paramref name="ev"/> unconditionally, producing the parked list for a
/// resolved handle that the atomic registration otherwise makes impossible.
/// </summary>
/// <param name="ev">The event to park.</param>
internal void ForceParkForTests(MxEvent ev)
{
lock (_parkGate)
{
var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev);
if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue))
{
queue = new Queue<ParkedEvent>();
_parkedByItemHandle[ev.ItemHandle] = queue;
}
queue.Enqueue(entry);
_parkOrder.Enqueue(entry);
Volatile.Write(ref _heldCount, _heldCount + 1);
}
}
/// <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.
@@ -230,78 +329,141 @@ internal sealed class SubscriptionRegistry
/// <summary>Gets the number of events currently held (parked or awaiting replay).</summary>
public int ParkedEventCount => Volatile.Read(ref _heldCount);
/// <summary>Gets the current gateway-session number (bumped by <see cref="StartNewSession"/>).</summary>
public int CurrentSession => Volatile.Read(ref _session);
/// <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"/>.
/// A new gateway session replaced the old one (reconnect). Discards every held event,
/// invalidates the reverse map — subscriptions are KEPT, but their old-session handles no
/// longer resolve, so new-session events park until the replay rebinds each subscription —
/// and bumps the session number that fences old pumps and straddling subscribes. NOT for the
/// same-session replay-gap re-snapshot, whose handles stay valid until rebound.
/// </summary>
/// <returns>The new epoch.</returns>
public int StartNewEpoch()
/// <returns>The new session number.</returns>
public int StartNewSession()
{
lock (_parkGate)
{
_parkedByItemHandle.Clear();
_ready.Clear();
Volatile.Write(ref _heldCount, 0);
return ++_epoch;
_subscribersByItemHandle.Clear();
DiscardAllHeldLocked();
// Volatile: the dispatch fast path reads it lock-free so a dying pump of the old session
// can never resolve an old-session event through the map once the new session exists.
Volatile.Write(ref _session, _session + 1);
return _session;
}
}
/// <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) —
/// not later: while any window is open, events for unknown handles are parked, not dropped.
/// not later: while a 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()
{
long sequence;
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);
sequence = ++_windowSequence;
_openWindows.Add(sequence);
Volatile.Write(ref _pendingSubscribes, _openWindows.Count);
}
return new PendingSubscribeScope(this);
return new PendingSubscribeScope(this, sequence);
}
private void EndPendingSubscribe()
private void EndPendingSubscribe(long sequence)
{
lock (_parkGate)
{
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();
_openWindows.Remove(sequence);
Volatile.Write(ref _pendingSubscribes, _openWindows.Count);
TrimUnclaimableLocked();
}
}
private void DiscardParkedLocked()
/// <summary>
/// Discard parked events no open window can claim: those stamped before the oldest open
/// window's sequence (all of them once no window is open). Promoted (ready) events are kept —
/// they belong to a registered handle and await replay. Amortised O(1) per parked event.
/// </summary>
private void TrimUnclaimableLocked()
{
if (_parkedByItemHandle.Count == 0) return;
var parked = 0;
foreach (var list in _parkedByItemHandle.Values) parked += list.Count;
var oldestOpen = _openWindows.Count == 0 ? long.MaxValue : _openWindows.Min;
while (_parkOrder.TryPeek(out var head) && head.Stamp < oldestOpen)
{
_parkOrder.Dequeue();
if (head.Gone) continue; // already promoted or purged
// Per-handle queues are in park order too, so a live head of the global order is the
// head of its handle's queue.
var queue = _parkedByItemHandle[head.ItemHandle];
queue.Dequeue();
head.Gone = true;
if (queue.Count == 0) _parkedByItemHandle.Remove(head.ItemHandle);
Volatile.Write(ref _heldCount, _heldCount - 1);
}
}
private void DiscardAllHeldLocked()
{
foreach (var parked in _parkOrder) parked.Gone = true;
_parkOrder.Clear();
_parkedByItemHandle.Clear();
Volatile.Write(ref _heldCount, _heldCount - parked);
_ready.Clear();
Volatile.Write(ref _heldCount, 0);
}
/// <summary>
/// Drop parked events for handles that were just unsubscribed (#859 second review, I-3
/// interim). A trailing event for a removed handle can be parked during another subscribe's
/// window; if MXAccess reuses that handle number for the other subscribe's item, promotion
/// would hand the old item's value to the new tag. Call when <c>UnsubscribeBulk</c> for
/// <paramref name="itemHandles"/> completes. It cannot catch an event still in flight
/// after that, nor tell a genuine early value for a reused number from a trailing one (it
/// drops both — a timeout is safer than another tag's value); a complete fix needs a
/// worker sequence on the SubscribeBulk reply (mxaccessgw, cross-repo).
/// </summary>
/// <param name="itemHandles">The handles whose unsubscribe completed.</param>
public void PurgeParked(IReadOnlyList<int> itemHandles)
{
if (Volatile.Read(ref _heldCount) == 0 || itemHandles.Count == 0) return;
lock (_parkGate)
{
foreach (var handle in itemHandles)
{
if (!_parkedByItemHandle.Remove(handle, out var queue)) continue;
foreach (var parked in queue) parked.Gone = true;
Volatile.Write(ref _heldCount, _heldCount - queue.Count);
}
}
}
/// <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
/// 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).
/// unknown while a subscribe is in flight — park the event for replay. Also hands back the
/// promoted events that must be dispatched BEFORE this one (one gate acquisition per event
/// while anything is held).
/// </summary>
/// <remarks>
/// ⚠️ The lock-free fast path relies on a single-dispatcher invariant: only the ONE live
/// <see cref="EventPump"/>'s dispatch thread ever parks (increments <c>_heldCount</c>), and it
/// is the thread calling this. So when it reads <c>_heldCount == 0</c>, no older event for
/// this handle can be waiting anywhere — it would have to have been parked by this same
/// thread earlier — and resolving through the map cannot overtake one. Registrations only
/// move events from parked to ready (count unchanged) and the dispatch thread only removes
/// them. A second concurrently-dispatching pump on one registry would break this.
/// </remarks>
/// <param name="ev">The data-change event.</param>
/// <param name="epoch">The calling pump's epoch (<see cref="StartNewEpoch"/>).</param>
/// <param name="session">The calling pump's session (<see cref="CurrentSession"/> at its creation).</param>
/// <param name="subscribers">The subscribers when the outcome is <see cref="EarlyEventOutcome.Resolved"/>.</param>
/// <param name="replayFirst">Promoted early events to dispatch before this one, in order.</param>
/// <returns>What happened to the event.</returns>
public EarlyEventOutcome ResolveOrPark(
MxEvent ev, int epoch, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers)
MxEvent ev, int session,
out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers,
out IReadOnlyList<MxEvent> replayFirst)
{
// 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)
replayFirst = [];
if (Volatile.Read(ref _heldCount) == 0 && Volatile.Read(ref _session) == session)
{
subscribers = ResolveSubscribers(ev.ItemHandle);
if (subscribers.Count > 0) return EarlyEventOutcome.Resolved;
@@ -310,7 +472,7 @@ internal sealed class SubscriptionRegistry
lock (_parkGate)
{
if (epoch != _epoch)
if (session != _session)
{
subscribers = [];
return EarlyEventOutcome.Stale; // a dying pump of an earlier session
@@ -319,28 +481,34 @@ internal sealed class SubscriptionRegistry
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))
// Defensive (M-1): registration promotes atomically with its map writes, so a
// resolved handle should hold no parked list. If one ever does, promote it now
// (fail safe) — parking this LIVE event behind it would lose it to a later trim if
// no registration ever claimed the list.
if (_parkedByItemHandle.Remove(ev.ItemHandle, out var stranded))
{
pending.Add(ev);
Volatile.Write(ref _heldCount, _heldCount + 1);
subscribers = [];
return EarlyEventOutcome.Parked;
foreach (var parked in stranded)
{
parked.Gone = true;
_ready.Enqueue(parked.Event);
}
}
replayFirst = TakeReadyLocked();
return EarlyEventOutcome.Resolved;
}
replayFirst = TakeReadyLocked();
if (_pendingSubscribes == 0) return EarlyEventOutcome.Stale;
if (_heldCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow;
if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var list))
var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev);
if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue))
{
list = [];
_parkedByItemHandle[ev.ItemHandle] = list;
queue = new Queue<ParkedEvent>();
_parkedByItemHandle[ev.ItemHandle] = queue;
}
list.Add(ev);
queue.Enqueue(entry);
_parkOrder.Enqueue(entry);
Volatile.Write(ref _heldCount, _heldCount + 1);
return EarlyEventOutcome.Parked;
}
@@ -348,23 +516,28 @@ internal sealed class SubscriptionRegistry
/// <summary>
/// Dequeue every event a registration promoted, in arrival order. Called only from the
/// EventPump's single dispatch thread. A pump of an earlier epoch gets nothing.
/// EventPump's single dispatch thread (the wake path). A pump of an earlier session gets nothing.
/// </summary>
/// <param name="epoch">The calling pump's epoch.</param>
/// <param name="session">The calling pump's session.</param>
/// <returns>The promoted events; empty when none are waiting.</returns>
public IReadOnlyList<MxEvent> TakeReadyEvents(int epoch)
public IReadOnlyList<MxEvent> TakeReadyEvents(int session)
{
if (Volatile.Read(ref _heldCount) == 0) return [];
lock (_parkGate)
{
if (epoch != _epoch || _ready.Count == 0) return [];
var taken = _ready.ToArray();
_ready.Clear();
Volatile.Write(ref _heldCount, _heldCount - taken.Length);
return taken;
return session != _session ? [] : TakeReadyLocked();
}
}
private IReadOnlyList<MxEvent> TakeReadyLocked()
{
if (_ready.Count == 0) return [];
var taken = _ready.ToArray();
_ready.Clear();
Volatile.Write(ref _heldCount, _heldCount - taken.Length);
return taken;
}
private bool PromoteParkedEventsLocked(IReadOnlyList<TagBinding> bindings)
{
if (_parkedByItemHandle.Count == 0) return false;
@@ -372,21 +545,37 @@ internal sealed class SubscriptionRegistry
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); // stays counted in _heldCount until replayed
if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var queue)) continue;
foreach (var parked in queue)
{
parked.Gone = true; // stays counted in _heldCount until replayed
_ready.Enqueue(parked.Event);
}
promoted = true;
}
return promoted;
}
private sealed class PendingSubscribeScope(SubscriptionRegistry owner) : IDisposable
private sealed class ParkedEvent(long stamp, int itemHandle, MxEvent ev)
{
/// <summary>Gets the newest window sequence when the event was parked.</summary>
public long Stamp { get; } = stamp;
/// <summary>Gets the event's item handle.</summary>
public int ItemHandle { get; } = itemHandle;
/// <summary>Gets the event.</summary>
public MxEvent Event { get; } = ev;
/// <summary>Gets or sets whether the event has left the parked set (promoted, purged or trimmed).</summary>
public bool Gone { get; set; }
}
private sealed class PendingSubscribeScope(SubscriptionRegistry owner, long sequence) : IDisposable
{
private int _disposed;
/// <summary>Closes this subscribe's early-event window (once).</summary>
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe();
if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe(sequence);
}
}
@@ -404,50 +593,6 @@ internal sealed class SubscriptionRegistry
public IReadOnlyList<(long SubscriptionId, IReadOnlyList<TagBinding> Bindings)> SnapshotEntries() =>
[.. _bySubscriptionId.Values.Select(entry => (entry.SubscriptionId, entry.Bindings))];
/// <summary>
/// Replace an existing subscription's bindings with the item handles a post-reconnect
/// SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new
/// handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the
/// subscription id is unknown (it was unsubscribed during the reconnect window).
/// </summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="newBindings">The new tag bindings after reconnection.</param>
public void Rebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings)
{
bool promoted;
lock (_parkGate) // map writes + promotion atomic — see Register
{
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)
{
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;
}
_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);
}
if (promoted) ParkedEventsReady?.Invoke();
}
/// <summary>
/// Per-subscription bookkeeping. <see cref="FullRefByItemHandle"/> is an index
/// over <see cref="Bindings"/> keyed by item handle so <c>ResolveSubscribers</c>
@@ -463,14 +608,18 @@ internal sealed class SubscriptionRegistry
public IReadOnlyList<TagBinding> Bindings { get; }
/// <summary>Gets the index of full references by item handle.</summary>
public IReadOnlyDictionary<int, string> FullRefByItemHandle { get; }
/// <summary>Gets the gateway session whose handles <see cref="Bindings"/> carries.</summary>
public int Session { get; }
/// <summary>Initializes a new subscription entry.</summary>
/// <param name="subscriptionId">The subscription identifier.</param>
/// <param name="bindings">The tag bindings for the subscription.</param>
public SubscriptionEntry(long subscriptionId, IReadOnlyList<TagBinding> bindings)
/// <param name="session">The session the bindings' handles were issued by.</param>
public SubscriptionEntry(long subscriptionId, IReadOnlyList<TagBinding> bindings, int session)
{
SubscriptionId = subscriptionId;
Bindings = bindings;
Session = session;
var index = new Dictionary<int, string>(bindings.Count);
foreach (var binding in bindings)
{