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:
@@ -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`
|
is bounded: `statuses` is an enrichment field, and `Hresult`/`ProtocolStatus`
|
||||||
stay acceptance-only regardless of which completion (if any) was attributed.
|
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
|
## Live alarm watch-list replacement
|
||||||
|
|
||||||
An auto-mode `SubscribeAlarms` with `fallback_enabled = true` always builds a
|
An auto-mode `SubscribeAlarms` with `fallback_enabled = true` always builds a
|
||||||
|
|||||||
@@ -333,6 +333,76 @@ public sealed class AlarmAckWriterTests
|
|||||||
Assert.Empty(server.RemovedItemHandles);
|
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]
|
[Fact]
|
||||||
public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne()
|
public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne()
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -183,6 +183,39 @@ public sealed class MxAccessWriteCompletionCacheTests
|
|||||||
Assert.Equal(77, Assert.Single(statuses).Detail);
|
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)
|
private static RepeatedField<MxStatusProxy> BuildStatuses(int detail)
|
||||||
{
|
{
|
||||||
return new RepeatedField<MxStatusProxy>
|
return new RepeatedField<MxStatusProxy>
|
||||||
|
|||||||
@@ -303,6 +303,9 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
|
|||||||
try { mx.Unregister(serverHandle); } catch { /* best effort */ }
|
try { mx.Unregister(serverHandle); } catch { /* best effort */ }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The registration is gone, so no completion can arrive for its handles.
|
||||||
|
completionCache.RemoveServer(serverHandle);
|
||||||
|
|
||||||
itemHandles.Clear();
|
itemHandles.Clear();
|
||||||
itemLru.Clear();
|
itemLru.Clear();
|
||||||
server = null;
|
server = null;
|
||||||
@@ -335,6 +338,10 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
|
|||||||
itemLru.Remove(node);
|
itemLru.Remove(node);
|
||||||
try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ }
|
try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ }
|
||||||
try { mx.RemoveItem(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()
|
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(
|
private static long CreateItemKey(
|
||||||
int serverHandle,
|
int serverHandle,
|
||||||
int itemHandle)
|
int itemHandle)
|
||||||
|
|||||||
Reference in New Issue
Block a user