fix(worker): review follow-up — ghost cache entries, ReadBulk dedupe/pumping, fault hygiene
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m15s
ci / portable (push) Failing after 4m19s
ci / windows-x86 (push) Successful in 6m16s

- AddItem/AddItem2/AddBufferedItem drop stale cache entries for a reused handle id
- ReadBulk: dedupe handles within a batch (advise/teardown once, fan out result);
  pump every 16 items in add/advise and teardown phases
- alarm poll stops without recording a fault once StaRuntime.LoopFault is set
- StaRuntime skips the loop-fault handler during a requested shutdown
- StaHung loop fault diagnostic carries type + HRESULT only
- gateway.md documents ReadBulk's shared deadline
This commit is contained in:
Joseph Doherty
2026-10-22 08:23:05 -04:00
parent 4ea3672b1e
commit 38fbbde191
9 changed files with 375 additions and 21 deletions
@@ -242,6 +242,7 @@ public sealed class MxAccessSession : IDisposable
ThrowIfDisposed();
int itemHandle = mxAccessServer.AddItem(serverHandle, itemDefinition);
EvictStaleHandleState(serverHandle, itemHandle);
handleRegistry.RegisterItemHandle(
serverHandle,
itemHandle,
@@ -265,6 +266,7 @@ public sealed class MxAccessSession : IDisposable
ThrowIfDisposed();
int itemHandle = mxAccessServer.AddItem2(serverHandle, itemDefinition, itemContext);
EvictStaleHandleState(serverHandle, itemHandle);
handleRegistry.RegisterItemHandle(
serverHandle,
itemHandle,
@@ -275,6 +277,30 @@ public sealed class MxAccessSession : IDisposable
return itemHandle;
}
/// <summary>
/// Drops cached per-handle state left over from a previous lifetime of
/// a handle id MXAccess just handed out again. RemoveItem and Unregister
/// already evict, but an OnDataChange or OnWriteComplete that was
/// queued on the STA when they ran is dispatched afterwards (the loop
/// pumps between work items) and recreates the entry. Without this, a
/// reused id would let ReadBulk's cached fast path serve that stale
/// value as <c>was_cached = true</c>. A handle the registry still
/// tracks is a live item (MXAccess returned an existing handle), so its
/// current entries are kept.
/// </summary>
private void EvictStaleHandleState(
int serverHandle,
int itemHandle)
{
if (handleRegistry.ContainsItemHandle(serverHandle, itemHandle))
{
return;
}
valueCache.Remove(serverHandle, itemHandle);
writeCompletionCache.Remove(serverHandle, itemHandle);
}
/// <summary>Removes an item from an MXAccess server.</summary>
/// <param name="serverHandle">Handle returned by the worker.</param>
/// <param name="itemHandle">Handle returned by the worker.</param>
@@ -409,6 +435,7 @@ public sealed class MxAccessSession : IDisposable
ThrowIfDisposed();
int itemHandle = mxAccessServer.AddBufferedItem(serverHandle, itemDefinition, itemContext);
EvictStaleHandleState(serverHandle, itemHandle);
handleRegistry.RegisterItemHandle(
serverHandle,
itemHandle,
@@ -957,7 +984,7 @@ public sealed class MxAccessSession : IDisposable
}
finally
{
TearDownSnapshots(serverHandle, snapshots);
TearDownSnapshots(serverHandle, snapshots, pumpStep);
}
}
@@ -970,6 +997,17 @@ public sealed class MxAccessSession : IDisposable
return ordered;
}
/// <summary>
/// Number of snapshot items ReadBulk adds/advises (phase 2) or tears
/// down (phase 4) between pump steps. Those phases make one or two COM
/// calls per tag without waiting, so a large batch would otherwise hold
/// the STA with no pump and no activity mark for its whole length; a
/// pump every <see cref="ReadBulkPumpInterval"/> items keeps inbound
/// callbacks flowing and keeps the STA heartbeat fresh, so a big batch
/// cannot drift toward <c>HeartbeatStuckCeiling</c>.
/// </summary>
internal const int ReadBulkPumpInterval = 16;
private void ReadSnapshots(
int serverHandle,
List<SnapshotRead> snapshots,
@@ -978,16 +1016,36 @@ public sealed class MxAccessSession : IDisposable
Action pumpStep)
{
// Phase 2: reserve our own item handle and advise for every uncached
// tag. A per-tag AddItem/Advise failure fails only that tag.
// tag. A per-tag AddItem/Advise failure fails only that tag. When
// MXAccess hands back a handle an earlier tag in this batch already
// advised (a duplicate address), that tag shares the earlier tag's
// subscription instead of advising the handle a second time, and gets
// the same outcome fanned out to it after the wait.
List<SnapshotRead> waiting = new(snapshots.Count);
foreach (SnapshotRead snapshot in snapshots)
Dictionary<int, SnapshotRead> advisedByHandle = new();
List<SnapshotRead> sharing = new();
for (int position = 0; position < snapshots.Count; position++)
{
SnapshotRead snapshot = snapshots[position];
try
{
if (position > 0 && position % ReadBulkPumpInterval == 0)
{
pumpStep();
}
snapshot.ItemHandle = AddItem(serverHandle, snapshot.TagAddress);
if (advisedByHandle.TryGetValue(snapshot.ItemHandle, out SnapshotRead? owner))
{
snapshot.SharedWith = owner;
sharing.Add(snapshot);
continue;
}
snapshot.Baseline = valueCache.CurrentVersion(serverHandle, snapshot.ItemHandle);
Advise(serverHandle, snapshot.ItemHandle);
snapshot.Advised = true;
advisedByHandle[snapshot.ItemHandle] = snapshot;
waiting.Add(snapshot);
}
catch (Exception exception)
@@ -996,11 +1054,32 @@ public sealed class MxAccessSession : IDisposable
}
}
if (waiting.Count == 0)
if (waiting.Count > 0)
{
return;
WaitForSnapshots(serverHandle, waiting, results, timeout, pumpStep);
}
foreach (SnapshotRead duplicate in sharing)
{
BulkReadResult? ownerResult = results[duplicate.SharedWith!.Index];
if (ownerResult is null)
{
continue;
}
BulkReadResult fanned = ownerResult.Clone();
fanned.TagAddress = duplicate.TagAddress;
results[duplicate.Index] = fanned;
}
}
private void WaitForSnapshots(
int serverHandle,
List<SnapshotRead> waiting,
BulkReadResult?[] results,
TimeSpan timeout,
Action pumpStep)
{
// Phase 3: one shared deadline. Pump until every advised handle has
// seen its first OnDataChange since its baseline, or the deadline
// elapses.
@@ -1054,17 +1133,20 @@ public sealed class MxAccessSession : IDisposable
private void TearDownSnapshots(
int serverHandle,
List<SnapshotRead> snapshots)
List<SnapshotRead> snapshots,
Action pumpStep)
{
// Phase 4: snapshot teardown — best-effort, never re-throws. A handle
// is torn down once even if MXAccess handed the same id to two tags in
// the batch.
// Phase 4: snapshot teardown — best-effort, never re-throws. Each
// distinct handle is unadvised and removed exactly once, even when
// MXAccess handed the same id to several tags in the batch.
HashSet<int> unadvised = new();
HashSet<int> removed = new();
int calls = 0;
foreach (SnapshotRead snapshot in snapshots)
{
if (snapshot.Advised && unadvised.Add(snapshot.ItemHandle))
{
PumpEvery(ref calls, pumpStep);
try { UnAdvise(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
}
}
@@ -1073,11 +1155,22 @@ public sealed class MxAccessSession : IDisposable
{
if (snapshot.ItemHandle != 0 && removed.Add(snapshot.ItemHandle))
{
PumpEvery(ref calls, pumpStep);
try { RemoveItem(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
}
}
}
private static void PumpEvery(ref int calls, Action pumpStep)
{
if (calls > 0 && calls % ReadBulkPumpInterval == 0)
{
try { pumpStep(); } catch { /* swallow — teardown is best effort */ }
}
calls++;
}
private bool TryGetCachedReadFor(
int serverHandle,
string tagAddress,
@@ -1429,6 +1522,12 @@ public sealed class MxAccessSession : IDisposable
/// <summary>Gets or sets a value indicating whether Advise succeeded.</summary>
public bool Advised { get; set; }
/// <summary>
/// Gets or sets the earlier snapshot in the same batch whose
/// subscription this one shares (same item handle), or null.
/// </summary>
public SnapshotRead? SharedWith { get; set; }
}
private void ThrowIfDisposed()
@@ -465,6 +465,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
// observable on the IPC fault path instead of silently
// faulting this never-awaited task. The loop then stops —
// the subscription is dead.
if (staRuntime.LoopFault is not null)
{
// The STA loop itself died; its handler already recorded
// the StaHung fault. This failure is just the fast-fail
// InvokeAsync returns afterwards — recording it would add
// a second, mislabelled alarm-poll fault.
return;
}
consecutiveFailures++;
if (consecutiveFailures >= failureThreshold)
{
@@ -545,8 +554,13 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// </summary>
internal static WorkerFault CreateStaLoopFault(Exception exception)
{
string message =
$"Worker STA loop terminated unexpectedly: {exception.GetType().FullName}: {exception.Message}";
// Type and HRESULT only, never exception.Message — the same type-only
// rule every other worker failure line follows.
string message = string.Format(
System.Globalization.CultureInfo.InvariantCulture,
"Worker STA loop terminated unexpectedly: exceptionType={0} hresult=0x{1:X8}",
exception.GetType().FullName,
exception.HResult);
WorkerFault fault = new()
{
Category = WorkerFaultCategory.StaHung,