fix: release the alarm drain latch on escape; cancel before disposing the page poll timers

This commit is contained in:
Joseph Doherty
2026-08-16 05:40:27 -04:00
parent b9e9df4081
commit 445bb855e2
5 changed files with 306 additions and 22 deletions
@@ -1546,6 +1546,83 @@ public sealed class ScriptedAlarmEngineTests
probe.Count.ShouldBe(2, "the in-flight pass plus exactly one follow-up pass for the late push");
}
/// <summary>
/// Verifies the drain latch (<c>_drainScheduled</c>) is released even when an exception
/// escapes the drain loop, so a later push still starts a drainer. A latch left stuck true
/// stops evaluation for EVERY scripted alarm, permanently and silently.
/// </summary>
/// <remarks>
/// The escape is produced without touching the engine: the store's SaveAsync throws (which
/// <c>ReevaluateAsync</c> catches, as designed), and the engine logger is configured with a
/// Serilog <b>audit</b> sink that throws on that catch block's own "reevaluate failed"
/// Error() call. Audit sinks propagate by contract, so the throw leaves <c>ReevaluateAsync</c>
/// through its catch and unwinds <c>DrainDirtyAsync</c> — the only escape path the current
/// code actually has, and a real one for any deployment auditing its logs.
/// </remarks>
[Fact]
public async Task A_throw_escaping_the_drain_loop_still_releases_the_drain_latch()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
var probe = new CountingSink("EVALPROBE");
var logger = new LoggerConfiguration()
.WriteTo.Sink(probe)
.AuditTo.Sink(new ThrowingSink("reevaluate failed"))
.CreateLogger();
var store = new CountingAlarmStateStore();
using var eng = new ScriptedAlarmEngine(up, store, new ScriptLoggerFactory(logger), logger);
await eng.LoadAsync([
Alarm("A", """ctx.Logger.Information("EVALPROBE"); return (int)ctx.GetTag("Temp").Value > 100;"""),
], TestContext.Current.CancellationToken);
// LoadAsync itself evaluates + persists once (startup recovery) — count from zero.
probe.Reset();
store.ResetCount();
// Push #1: transition → SaveAsync throws → the catch's Error() hits the audit
// sink and throws → the exception unwinds out of the drain loop. Waiting on the
// save ATTEMPT (counted before the throw) rather than on the probe keeps the
// predicate-evaluation count below assertable.
store.ThrowOnNextSave = true;
up.Push("Temp", 150);
await WaitForAsync(() => store.SaveCount == 1, timeoutMs: 3000);
probe.Count.ShouldBe(1, "the faulted pass evaluated the predicate exactly once");
eng.GetState("A")!.Active.ShouldBe(AlarmActiveState.Inactive,
"persist-before-update: a failed save must leave the in-memory condition at the prior state");
// The pin. This push either starts a fresh drainer (latch already released) or is
// handed to the replacement drainer the unwinding pass starts (latch still held,
// dirty set non-empty) — the two orderings the race between "SaveAsync was called"
// and "the drain finished unwinding" produces, and both must evaluate it. With the
// latch stuck true, OnUpstreamChange only grows the dirty set, no predicate ever
// runs again, and the alarm silently never activates.
up.Push("Temp", 160);
await WaitForAsync(() => eng.GetState("A")!.Active == AlarmActiveState.Active, timeoutMs: 3000);
probe.Count.ShouldBe(2, "the push after the faulted drain must itself be evaluated");
}
/// <summary>
/// A Serilog sink that throws when a marker event is emitted. Registered through
/// <c>AuditTo</c> so the failure propagates to the caller instead of being swallowed by
/// Serilog's normal best-effort emit — the mechanism the drain-latch test uses to make an
/// exception escape a method that otherwise catches everything.
/// </summary>
private sealed class ThrowingSink : Serilog.Core.ILogEventSink
{
private readonly string _marker;
/// <summary>Initializes the sink with the marker substring that triggers the throw.</summary>
/// <param name="marker">Substring the message template must contain to throw.</param>
public ThrowingSink(string marker) => _marker = marker;
/// <summary>Throws if the event's message template carries the marker; otherwise ignores it.</summary>
/// <param name="logEvent">The Serilog event being emitted.</param>
public void Emit(Serilog.Events.LogEvent logEvent)
{
if (logEvent.MessageTemplate.Text.Contains(_marker, StringComparison.Ordinal))
throw new InvalidOperationException($"audit sink refused '{_marker}' (test)");
}
}
/// <summary>
/// A Serilog sink that counts the events whose message template contains a marker string.
/// Used to count predicate EVALUATIONS (the predicate emits the marker through
@@ -1598,6 +1675,12 @@ public sealed class ScriptedAlarmEngineTests
/// <summary>Gets a value indicating whether a SaveAsync call is currently blocked waiting on BlockNextSave.</summary>
public bool SaveInProgress { get; private set; }
/// <summary>
/// Gets or sets a value indicating whether the next SaveAsync throws instead of persisting.
/// Self-clearing, so one failed save can be followed by a healthy one.
/// </summary>
public bool ThrowOnNextSave { get; set; }
/// <summary>Loads an alarm condition state by ID from the inner store.</summary>
/// <param name="alarmId">The ID of the alarm condition state to load.</param>
/// <param name="ct">A cancellation token.</param>
@@ -1609,12 +1692,18 @@ public sealed class ScriptedAlarmEngineTests
public Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct)
=> _inner.LoadAllAsync(ct);
/// <summary>Counts the save, optionally blocking on the BlockNextSave gate, then delegates.</summary>
/// <summary>Counts the save, optionally throwing or blocking on the BlockNextSave gate, then delegates.</summary>
/// <param name="state">The alarm condition state to save.</param>
/// <param name="ct">A cancellation token.</param>
public async Task SaveAsync(AlarmConditionState state, CancellationToken ct)
{
Interlocked.Increment(ref _saveCount);
if (ThrowOnNextSave)
{
ThrowOnNextSave = false;
throw new InvalidOperationException("store unavailable (test)");
}
var gate = BlockNextSave;
if (gate is not null)
{