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

fix(alarms): replacement drainer starts on a fresh stack; honest bounds in the latch comments

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 05:57:05 -04:00
parent cd010c7165
commit c8a7014388
2 changed files with 150 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs
+32 -7
View File
@@ -623,12 +623,37 @@ public sealed class ScriptedAlarmEngine : IDisposable
// carrying dies with it — ReevaluateAsync's own catch is what normally makes
// that a logged non-event — but ids that queued BEHIND it are still in the
// set and would otherwise sit there until some later push happened to arrive,
// so hand the latch straight to a replacement drainer instead of releasing
// it. Handing over under the same lock keeps "exactly one drainer" true with
// no window a push could start a second one in. This cannot spin: every pass
// swaps the set out and clears it before evaluating, so a replacement finds
// work only if fresh pushes arrived, and a permanently-throwing evaluation
// costs one pass per push — the same rate as a healthy one.
// so hand the latch straight to a replacement drainer instead of releasing it.
//
// Exactly-one-drainer survives the handover because _drainScheduled never
// transiently CLEARS during one: deciding to hand over and clearing the flag
// are the same guarded read-modify-write, and the clear happens only in the
// branch that starts no replacement. The spawn itself is deliberately OUTSIDE
// the lock (nothing it does belongs under _dirtyLock) and is safe there
// precisely because the flag is still set — a push landing in that window only
// grows the set, which the replacement then drains.
//
// Two bounds, both honest:
// * CPU — every pass swaps the set out and clears it before evaluating, so a
// replacement finds work only if fresh pushes arrived. A permanently
// throwing evaluation therefore costs one pass per push, the same rate as
// a healthy one; it cannot spin against an idle upstream.
// * Stack — the replacement is QUEUED (Task.Run), not called. A replacement
// that is called runs INLINE for as long as it does not truly suspend, and
// its own escape then unwinds back into this same finally block, nesting a
// frame per throw with no ceiling — a StackOverflowException no catch can
// recover from. Whether it suspends first is not a property this method
// controls: _evalGate is already free when the escape is thrown (released
// in ReevaluateAsync's inner finally, before the outer catch's logging
// call), so a failure raised BEFORE the first awaited work — the clock
// read, anything synchronous ahead of the predicate — unwinds with no
// suspension at all. A failure raised at or after the predicate does not,
// because TimedScriptEvaluator hops to the pool on every evaluation;
// measured, that path holds depth flat even when called directly. Queuing
// makes the fresh stack unconditional instead of dependent on which throw
// shape happens to occur, at the price of one pool dispatch per faulted
// pass. Pinned by
// Sustained_synchronous_drain_failures_hand_over_without_deepening_the_stack.
if (!released)
{
bool restart;
@@ -638,7 +663,7 @@ public sealed class ScriptedAlarmEngine : IDisposable
if (!restart) _drainScheduled = false;
}
if (restart) TrackBackgroundTask(DrainDirtyAsync(ct));
if (restart) TrackBackgroundTask(Task.Run(() => DrainDirtyAsync(ct)));
}
}
}
tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs
+118
View File
@@ -1600,6 +1600,124 @@ public sealed class ScriptedAlarmEngineTests
probe.Count.ShouldBe(2, "the push after the faulted drain must itself be evaluated");
}
/// <summary>
/// Verifies a long chain of drain-latch handovers under sustained <b>synchronous</b> failure
/// neither stalls the engine nor deepens the native stack: each replacement drainer must
/// start on a FRESH stack rather than nesting inside the finally block that spawned it.
/// </summary>
/// <remarks>
/// <para><b>The escape has to be synchronous, and picking the wrong one makes this test
/// vacuous.</b> A drain pass whose failure comes from the STORE cannot nest: the predicate
/// runs first, and <c>TimedScriptEvaluator.RunAsync</c> hops to the pool
/// (<c>Task.Run(...).WaitAsync(timeout)</c>) on every evaluation, so the drainer has already
/// suspended and returned to its caller long before the throw. Measured: depth is a flat 36
/// across 200 store-throw handovers whether the replacement is queued or called. The escape
/// used here is therefore the injected <b>clock</b>, which <c>ReevaluateAsync</c> calls at
/// the top of each alarm's iteration — under a free <c>_evalGate</c>, before any awaited
/// work — so the throw, the catch, the audit sink's rethrow and the whole unwind happen
/// without a single suspension.</para>
/// <para>The chain is deterministic rather than raced: the clock re-dirties the alarm from
/// inside the failing call (a re-entrant upstream push, which only grows the dirty set
/// because the latch is already held), so every unwinding pass finds work and hands the
/// latch to a replacement — exactly <c>Handovers</c> times, with no dependence on thread
/// timing. Depth is sampled per call: flat when the handover goes through <c>Task.Run</c>,
/// climbing monotonically when the replacement is called inline.</para>
/// </remarks>
[Fact]
public async Task Sustained_synchronous_drain_failures_hand_over_without_deepening_the_stack()
{
const int Handovers = 200;
var up = new FakeUpstream();
up.Set("Temp", 50);
var logger = new LoggerConfiguration()
.AuditTo.Sink(new ThrowingSink("reevaluate failed"))
.CreateLogger();
var clock = new RedirtyingThrowingClock(up, "Temp", Handovers);
using var eng = new ScriptedAlarmEngine(
up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger, clock.Now);
await eng.LoadAsync([
Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;"""),
], TestContext.Current.CancellationToken);
// Arm AFTER load: LoadAsync's startup-recovery pass reads the clock too, and it must not
// be the pass that throws.
clock.Arm();
// One push starts the chain; every failing read re-dirties and throws, so the chain
// self-sustains for exactly Handovers handovers and then stops.
up.Push("Temp", 150);
await WaitForAsync(() => clock.CallCount >= Handovers + 1, timeoutMs: 60000);
clock.CallCount.ShouldBe(Handovers + 1,
"every handover must produce exactly one further evaluation pass — no stall, no duplication");
(clock.MaxDepth - clock.FirstDepth).ShouldBeLessThan(200,
$"a queued handover starts on a fresh stack, so depth must stay flat across {Handovers} " +
$"handovers (first={clock.FirstDepth}, max={clock.MaxDepth})");
// Liveness: the latch was released at the end of the chain, so the engine still works.
clock.Disarm();
up.Push("Temp", 170);
await WaitForAsync(() => eng.GetState("A")!.Active == AlarmActiveState.Active, timeoutMs: 5000);
}
/// <summary>
/// A clock that, once armed, re-dirties an alarm by pushing to the upstream and then throws —
/// driving a deterministic chain of drain-latch handovers whose every step is synchronous —
/// while sampling the native stack depth of each call.
/// </summary>
private sealed class RedirtyingThrowingClock
{
private readonly FakeUpstream _up;
private readonly string _path;
private readonly int _redirtyLimit;
private readonly object _gate = new();
private volatile bool _armed;
private int _calls;
private int _firstDepth;
private int _maxDepth;
/// <summary>Initializes the clock with the upstream to re-dirty and the handover count to drive.</summary>
/// <param name="up">Upstream fake the failing call re-pushes to.</param>
/// <param name="path">Tag path to re-push, which must be one the alarm reads.</param>
/// <param name="redirtyLimit">Number of calls that re-dirty before the chain is allowed to end.</param>
public RedirtyingThrowingClock(FakeUpstream up, string path, int redirtyLimit)
{ _up = up; _path = path; _redirtyLimit = redirtyLimit; }
/// <summary>Gets the number of clock reads taken while armed.</summary>
public int CallCount => Volatile.Read(ref _calls);
/// <summary>Gets the stack depth sampled on the first armed read.</summary>
public int FirstDepth => Volatile.Read(ref _firstDepth);
/// <summary>Gets the deepest stack sampled across all armed reads.</summary>
public int MaxDepth => Volatile.Read(ref _maxDepth);
/// <summary>Starts failing (and re-dirtying) on every read.</summary>
public void Arm() => _armed = true;
/// <summary>Stops failing, so the clock behaves normally again.</summary>
public void Disarm() => _armed = false;
/// <summary>Samples the stack depth, re-dirties the alarm, and throws — or returns the time when disarmed.</summary>
public DateTime Now()
{
if (!_armed) return DateTime.UtcNow;
var n = Interlocked.Increment(ref _calls);
var depth = new System.Diagnostics.StackTrace(false).FrameCount;
lock (_gate)
{
if (n == 1) _firstDepth = depth;
if (depth > _maxDepth) _maxDepth = depth;
}
// Re-dirty from inside the failing pass: the latch is held by the drainer that
// called us, so this only grows the dirty set — which is exactly the condition
// that makes the unwinding drainer hand the latch over instead of releasing it.
if (n <= _redirtyLimit) _up.Push(_path, 150);
throw new InvalidOperationException("clock unavailable (test)");
}
}
/// <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