fix(worker): forget write-completion entries for dropped ack item handles

Add MxAccessWriteCompletionCache.Remove and RemoveServer and call them
from AlarmAckWriter when it drops an item (dirty re-add, LRU eviction,
failed AddItem/Advise/Write) and when it tears down its registration, so
the cache no longer keeps one entry per dropped handle for the life of
the worker.

Fixes #165

Claude-Session: https://claude.ai/code/session_016J6isfcgqy4SB2RK3ZvFvT
This commit is contained in:
Joseph Doherty
2026-10-20 07:09:59 -04:00
parent 219611c821
commit f13950e453
5 changed files with 172 additions and 0 deletions
+11
View File
@@ -351,6 +351,17 @@ native client correlating by handle has the identical ambiguity). The exposure
is bounded: `statuses` is an enrichment field, and `Hresult`/`ProtocolStatus`
stay acceptance-only regardless of which completion (if any) was attributed.
**The alarm ack writer narrows the window for acknowledgements.**
`AlarmAckWriter` registers its own MXAccess server with its own
`MxAccessWriteCompletionCache` and reuses one advised `AckMsg` item per alarm
(least-recently-used, capped at 256). After an ack ends unconfirmed it marks the
item dirty, and the next ack to that alarm removes and re-adds the item so the
write waits on a fresh item handle the stale completion cannot match. Every
drop of an item (dirty re-add, eviction, or a failed `AddItem`/`Advise`/`Write`)
also removes that handle's completion entry, and tearing down the registration
removes every entry for its server handle, so the cache does not grow by one
entry per dropped handle for the life of the worker.
## Live alarm watch-list replacement
An auto-mode `SubscribeAlarms` with `fallback_enabled = true` always builds a
@@ -333,6 +333,76 @@ public sealed class AlarmAckWriterTests
Assert.Empty(server.RemovedItemHandles);
}
[Fact]
public void Acknowledge_AfterDirtyReAdd_ForgetsDroppedHandleCompletion()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(20), "MxGateway");
writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, Unconfirmed
cache.Record(7, 100, Rows(MxStatusCategory.Ok, 0)); // its late completion
Assert.Equal(1, cache.Count);
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
AlarmAckResult second = writer.Acknowledge("T.A.AckMsg", "two", Alice()); // drops 100, adds 101
Assert.Equal(AlarmAckOutcome.Applied, second.Outcome);
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 100));
Assert.Equal(1UL, cache.CurrentVersion(7, 101));
}
[Fact]
public void Acknowledge_PastCacheCap_ForgetsEvictedHandleCompletion()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway", maxCachedItems: 2);
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
Assert.Equal(2, cache.Count);
writer.Acknowledge("C.X.AckMsg", "3", Alice()); // handle 102, evicts A (100)
Assert.Equal(2, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 100));
}
[Fact]
public void Dispose_ForgetsServerCompletions()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("A.X.AckMsg", "1", Alice());
writer.Acknowledge("B.X.AckMsg", "2", Alice());
cache.Record(8, 100, Rows(MxStatusCategory.Ok, 0)); // another server's entry survives
writer.Dispose();
Assert.Equal(1, cache.Count);
Assert.Equal(1UL, cache.CurrentVersion(8, 100));
}
[Fact]
public void Acknowledge_WhenWriteDisconnected_ForgetsServerCompletions()
{
RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100 recorded
server.WriteException = new COMException("x", unchecked((int)0x80010108));
writer.Acknowledge("A.X.AckMsg", "2", Alice());
Assert.Contains("Unregister", server.Operations);
Assert.Equal(0, cache.Count);
}
[Fact]
public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne()
{
@@ -183,6 +183,39 @@ public sealed class MxAccessWriteCompletionCacheTests
Assert.Equal(77, Assert.Single(statuses).Detail);
}
/// <summary>Verifies that Remove forgets only the given handle pair.</summary>
[Fact]
public void Remove_ForgetsOnlyThatHandlePair()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, 22, BuildStatuses(detail: 2));
cache.Remove(7, 21);
cache.Remove(7, 99); // unknown pair is a no-op
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 21));
Assert.Equal(1UL, cache.CurrentVersion(7, 22));
}
/// <summary>Verifies that RemoveServer forgets every pair of that server and no other.</summary>
[Fact]
public void RemoveServer_ForgetsOnlyThatServersPairs()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, -5, BuildStatuses(detail: 2));
cache.Record(8, 21, BuildStatuses(detail: 3));
cache.Record(-7, 21, BuildStatuses(detail: 4));
cache.RemoveServer(7);
Assert.Equal(2, cache.Count);
Assert.Equal(1UL, cache.CurrentVersion(8, 21));
Assert.Equal(1UL, cache.CurrentVersion(-7, 21));
}
private static RepeatedField<MxStatusProxy> BuildStatuses(int detail)
{
return new RepeatedField<MxStatusProxy>
@@ -303,6 +303,9 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
try { mx.Unregister(serverHandle); } catch { /* best effort */ }
}
// The registration is gone, so no completion can arrive for its handles.
completionCache.RemoveServer(serverHandle);
itemHandles.Clear();
itemLru.Clear();
server = null;
@@ -335,6 +338,10 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
itemLru.Remove(node);
try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ }
try { mx.RemoveItem(serverHandle, node.Value.Handle); } catch { /* best effort */ }
// Without this the completion cache keeps one entry per dropped handle
// (each Unconfirmed re-add and each eviction) for the worker's life.
completionCache.Remove(serverHandle, node.Value.Handle);
}
private IMxAccessServer EnsureRegistered()
@@ -184,6 +184,57 @@ public sealed class MxAccessWriteCompletionCache
}
}
/// <summary>
/// Forgets the completion entry of a handle pair whose MXAccess item was
/// removed, so a long-lived writer that drops and re-adds items does not
/// keep one entry per dropped handle for the life of the worker.
/// </summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle that was removed.</param>
public void Remove(
int serverHandle,
int itemHandle)
{
lock (syncRoot)
{
entries.Remove(CreateItemKey(serverHandle, itemHandle));
}
}
/// <summary>Forgets every completion entry recorded under a server handle that was unregistered.</summary>
/// <param name="serverHandle">MXAccess server handle that was unregistered.</param>
public void RemoveServer(int serverHandle)
{
lock (syncRoot)
{
List<long> stale = new();
foreach (long key in entries.Keys)
{
if ((int)(key >> 32) == serverHandle)
{
stale.Add(key);
}
}
foreach (long key in stale)
{
entries.Remove(key);
}
}
}
/// <summary>Gets the number of handle pairs with a recorded completion (test seam).</summary>
internal int Count
{
get
{
lock (syncRoot)
{
return entries.Count;
}
}
}
private static long CreateItemKey(
int serverHandle,
int itemHandle)