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

fix(calc): evaluate off the mux callback thread and out from under the state lock (Driver.Calculation-005)

Browse Source 
OnDependencyValue ran every change-triggered dependent's script INLINE, under _lock,
via TimedScriptEvaluator.RunAsync(...).GetAwaiter().GetResult() — on the Akka dispatcher
thread of Runtime's DependencyConsumerMuxAdapter, for up to RunTimeout (2 s default) per
arrival, with the state lock held for the whole run.

The callback now records the value, PREPARES the affected tags under the lock (definition
+ warm-up gate + dependency snapshot taken at trigger time) and queues them for a single
serialized evaluation pump. _lock is a short-held state lock again; a new _evalGate
serializes the pump against the timer ticks and is taken only by evaluation workers, so
no host thread can wait on a script. Lock order is always _evalGate -> _lock.

Semantics preserved: per-tag ordering (one FIFO pump), the warm-up gate BEFORE the
evaluator runs, dedup/Bad-transition/recovery publishes, and — because the dependency
snapshot is taken when the trigger fires rather than when the pump catches up — a burst
of arrivals still evaluates against the values that triggered it.
This commit is contained in:
Joseph Doherty
2026-08-10 08:02:44 -04:00
parent 94b8633e68
commit 235e06a875
2 changed files with 305 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
+211 -54
View File
@@ -28,6 +28,16 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
/// at least once, and equal results are deduped) and/or on a timer (one timer per interval group).
/// </para>
/// <para>
/// <b>Where scripts run (Driver.Calculation-005).</b> <see cref="OnDependencyValue"/> is called
/// directly and synchronously by <c>DependencyConsumerMuxAdapter</c> — on an Akka actor's own
/// dispatcher thread — so it does no evaluation itself: it records the value, <i>prepares</i> the
/// affected tags (definition + dependency snapshot, taken at trigger time) and hands them to a
/// single serialized <b>evaluation pump</b>. Script execution therefore never runs on the mux
/// callback thread and never runs under <see cref="_lock"/>; the pump and the timer ticks are
/// serialized against each other by <see cref="_evalGate"/> instead, which preserves the
/// one-evaluation-at-a-time ordering the state lock used to provide.
/// </para>
/// <para>
/// <b>Error semantics (mini-design §6).</b> An evaluator <c>Failure</c> publishes <b>Bad</b> quality
/// carrying the last-known value, once per Good→Bad transition and only after all deps have arrived,
/// and emits a <c>ScriptLogEntry</c> on the <c>script-logs</c> topic (via the injected
@@ -67,13 +77,27 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
// Reassigned by RebuildTagTable so a redeploy that adds/removes dependencies re-registers on the mux.
private IReadOnlyCollection<string> _dependencyRefs = Array.Empty<string>();
// Live evaluation state. Guarded by _lock: OnDependencyValue (actor thread) and timer ticks
// (pool threads) can both drive an evaluation, so all state mutation is serialized.
// Live evaluation state. Guarded by _lock: OnDependencyValue (the mux actor's thread), timer ticks
// and the evaluation pump (pool threads) all touch it, so state mutation is serialized. _lock is a
// SHORT-HELD state lock — it is taken for dictionary work only and is NEVER held across a script run
// (Driver.Calculation-005).
private readonly object _lock = new();
private readonly Dictionary<string, object?> _depValues = new(StringComparer.Ordinal);
private readonly HashSet<string> _arrivedDeps = new(StringComparer.Ordinal);
private readonly Dictionary<string, TagState> _stateByRawPath = new(StringComparer.Ordinal);
// Serializes script EXECUTION (and the result-apply that follows it) across the evaluation pump and
// the timer ticks — the mutual exclusion _lock used to provide while it was held across the run, minus
// the blocking. Taken ONLY by evaluation workers, so no host thread (mux callback, ReadAsync,
// reinit) can ever wait on a script. Lock order is always _evalGate → _lock, never the reverse.
private readonly object _evalGate = new();
// Change-triggered evaluations queued by OnDependencyValue and drained, in arrival order, by a single
// pump task. Both fields are guarded by _lock; _pumpScheduled stays true until the pump observes an
// empty queue under the lock, so an enqueue can never be lost against a pump that is exiting.
private readonly Queue<PendingEvaluation> _pending = new();
private bool _pumpScheduled;
private CalculationTimerScheduler? _timers;
private volatile bool _running;
private DateTime? _lastComputeUtc;
@@ -255,6 +279,10 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
var newOptions = ParseOptions(driverConfigJson);
lock (_lock)
{
// Queued evaluations were prepared against the OUTGOING tag table (old script source, old
// dependency set), so they are stale the moment the rebuild lands — drop them rather than
// publish one last pre-edit value after the delta was applied.
_pending.Clear();
RebuildTagTable(newOptions);
}
// Drop compiled scripts (IScriptCacheOwner contract — stops collectible ALCs accreting AND forces a
@@ -270,6 +298,8 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
{
_running = false;
StopTimers();
// The pump exits at its next iteration; an evaluation already in flight is bounded by RunTimeout.
lock (_lock) _pending.Clear();
return Task.CompletedTask;
}
@@ -300,16 +330,24 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
public void OnDependencyValue(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
{
if (!_running) return;
List<string>? dependents;
lock (_lock)
{
_depValues[rawPath] = value;
_arrivedDeps.Add(rawPath);
if (!_changeDependents.TryGetValue(rawPath, out dependents) || dependents.Count == 0) return;
// Evaluate each change-triggered dependent inline under the lock (the compile cache makes this
// a bounded method invocation — the same "fast enough to run inline" contract as the VT evaluator).
if (!_changeDependents.TryGetValue(rawPath, out var dependents) || dependents.Count == 0) return;
// QUEUE each change-triggered dependent — never evaluate here (Driver.Calculation-005). This
// runs on the caller's thread, which for the live path is the Akka dispatcher thread of
// Runtime's DependencyConsumerMuxAdapter, and a script run is bounded only by RunTimeout.
// Everything under this lock is dictionary work; the script runs on the pump.
var queued = false;
foreach (var calcRawPath in dependents)
EvaluateTag(calcRawPath, timestampUtc);
{
if (PrepareEvaluation(calcRawPath, timestampUtc) is not { } prepared) continue;
_pending.Enqueue(prepared);
queued = true;
}
if (queued) SchedulePump();
}
}
@@ -362,79 +400,188 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
// ---- evaluation ----
/// <summary>Timer-trigger entry point — evaluates a single calc tag. Guards the lock itself (called
/// from a pool thread, unlike the change path which is already under the lock).</summary>
private void EvaluateTagLocked(string rawPath)
/// <summary>One prepared evaluation: the calc tag's definition, the dependency snapshot taken when its
/// trigger fired, and that trigger's timestamp. Built under <see cref="_lock"/> by
/// <see cref="PrepareEvaluation"/> and executed outside it, so the script itself needs no shared state.</summary>
/// <param name="Definition">The calc tag being evaluated.</param>
/// <param name="Dependencies">Its declared reads' values, as of the trigger.</param>
/// <param name="TimestampUtc">The trigger's timestamp, stamped on whatever is published.</param>
private sealed record PendingEvaluation(
CalculationTagDefinition Definition,
IReadOnlyDictionary<string, object?> Dependencies,
DateTime TimestampUtc);
/// <summary>
/// Prepare one calc tag's evaluation: resolve the definition, apply the change-gate, and snapshot
/// the tag's declared dependency values. MUST be called under <see cref="_lock"/>. Returns
/// <see langword="null"/> when the tag must not evaluate — unknown, no resolved script source
/// (logged at init), or still warming up.
/// <para>Change-gate (mini-design §4): publish nothing until every declared dep has arrived at
/// least once. This MUST hold BEFORE the evaluator runs, not just before the publish: a
/// partially-warmed-up snapshot is missing entries for the not-yet-arrived deps, and
/// <c>VirtualTagContext.GetTag</c> returns a Bad snapshot (<c>.Value == null</c>) for those — so the
/// common <c>(double)ctx.GetTag("x").Value</c> idiom throws <c>NullReferenceException</c> on every
/// warm-up tick for any multi-dependency script whose deps don't all arrive in the same tick.
/// Evaluating anyway (and discarding the result afterwards) used to turn ordinary warm-up into a
/// spurious "script threw" script-log Warning on every partial arrival, plus a wasted
/// Roslyn/compiled-cache invocation. Applying it here rather than on the pump is equivalent:
/// <see cref="_arrivedDeps"/> only ever grows, and the arrival that completes the set queues its own
/// evaluation.</para>
/// <para>Snapshotting the dependency values HERE — at trigger time — is what keeps the queued path
/// semantically identical to the old inline one: a burst of arrivals evaluates against the values
/// that triggered it rather than collapsing onto whatever happened to be current when the pump
/// caught up.</para>
/// </summary>
/// <param name="rawPath">The calc tag to prepare.</param>
/// <param name="timestampUtc">The trigger's timestamp.</param>
/// <returns>The prepared evaluation, or null when the tag must not evaluate.</returns>
private PendingEvaluation? PrepareEvaluation(string rawPath, DateTime timestampUtc)
{
if (!_running) return;
lock (_lock) EvaluateTag(rawPath, DateTime.UtcNow);
}
if (!_tagsByRawPath.TryGetValue(rawPath, out var def)) return null;
if (string.IsNullOrWhiteSpace(def.ScriptSource)) return null; // unresolved script — never computes
if (!def.DependencyRefs.All(_arrivedDeps.Contains)) return null; // still warming up — do not evaluate
/// <summary>Evaluate one calc tag and publish the outcome. MUST be called under <see cref="_lock"/>.</summary>
private void EvaluateTag(string rawPath, DateTime timestampUtc)
{
if (!_tagsByRawPath.TryGetValue(rawPath, out var def)) return;
if (string.IsNullOrWhiteSpace(def.ScriptSource)) return; // unresolved script — never computes (logged at init)
var st = _stateByRawPath[rawPath];
// Change-gate (mini-design §4): publish nothing until every declared dep has arrived at least once.
// MUST be checked BEFORE the evaluator runs, not just before the publish: a partially-warmed-up
// dependency snapshot is missing entries for the not-yet-arrived deps, and VirtualTagContext.GetTag
// returns a Bad snapshot (.Value == null) for those — so the common `(double)ctx.GetTag("x").Value`
// idiom throws NullReferenceException on every warm-up tick for any multi-dependency script whose
// deps don't all arrive in the same tick. Evaluating anyway (and discarding the result afterwards)
// used to turn ordinary warm-up into a spurious "script threw" script-log Warning on every partial
// arrival, plus a wasted Roslyn/compiled-cache invocation.
if (!def.DependencyRefs.All(_arrivedDeps.Contains)) return; // still warming up — do not evaluate yet
// Build the per-tag dependency snapshot (only its declared reads, current values).
var deps = new Dictionary<string, object?>(StringComparer.Ordinal);
foreach (var dep in def.DependencyRefs)
if (_depValues.TryGetValue(dep, out var v)) deps[dep] = v;
return new PendingEvaluation(def, deps, timestampUtc);
}
var result = _evaluator.Evaluate(rawPath, def.ScriptSource, deps);
/// <summary>Start the evaluation pump if it is not already running. MUST be called under
/// <see cref="_lock"/> — the flag and the queue are checked together, which is what makes "an enqueue
/// either starts the pump or is seen by a pump that has not yet exited" hold.</summary>
private void SchedulePump()
{
if (_pumpScheduled) return;
_pumpScheduled = true;
_ = Task.Run(RunEvaluationPump);
}
if (!result.Success)
/// <summary>Drain the queued change-triggered evaluations, in arrival order, on a pool thread. At most
/// one pump runs at a time, so per-tag evaluation ordering is preserved exactly as it was when the
/// change path evaluated inline under the state lock.</summary>
private void RunEvaluationPump()
{
while (true)
{
OnEvaluationFailed(def, st, result.Reason ?? "evaluator failure", timestampUtc);
return;
PendingEvaluation next;
lock (_lock)
{
if (!_running) _pending.Clear();
if (_pending.Count == 0)
{
_pumpScheduled = false;
return;
}
next = _pending.Dequeue();
}
// _evalGate (never _lock) is held across the run, so the pump and the timer ticks still
// evaluate one at a time while no host thread can block behind a script.
lock (_evalGate)
{
try
{
RunEvaluation(next);
}
catch (Exception ex)
{
// The evaluator never throws for script faults, so this is a defect path only — it
// must not kill the pump and strand every queued evaluation behind it.
_logger.LogError(ex, "Calculation {Driver}: evaluation of {RawPath} failed",
_driverInstanceId, next.Definition.RawPath);
}
}
}
}
/// <summary>Timer-trigger entry point — prepares and evaluates a single calc tag on the scheduler's
/// pool thread (the scheduler's per-group in-flight guard is what bounds outstanding timer work).</summary>
private void EvaluateTagFromTimer(string rawPath)
{
if (!_running) return;
PendingEvaluation? prepared;
lock (_lock) prepared = PrepareEvaluation(rawPath, DateTime.UtcNow);
if (prepared is null) return;
lock (_evalGate) RunEvaluation(prepared);
}
/// <summary>Run one prepared evaluation and publish the outcome. MUST be called under
/// <see cref="_evalGate"/> and NOT under <see cref="_lock"/>: the script run is the whole point of the
/// split, and the script log + <see cref="OnDataChange"/> fan-out both invoke subscriber code.</summary>
/// <param name="pending">The prepared evaluation.</param>
private void RunEvaluation(PendingEvaluation pending)
{
var def = pending.Definition;
var result = _evaluator.Evaluate(def.RawPath, def.ScriptSource, pending.Dependencies);
string? failureReason = null;
(object? Value, uint StatusCode)? toPublish;
lock (_lock)
{
// A ReinitializeAsync can drop the tag while its script is running; its result is then stale
// by definition and there is no state left to record it against.
if (!_stateByRawPath.TryGetValue(def.RawPath, out var st)) return;
if (!result.Success)
{
failureReason = result.Reason ?? "evaluator failure";
toPublish = MarkEvaluationFailed(st, pending.TimestampUtc);
}
else
{
toPublish = MarkEvaluationSucceeded(st, result.Value, pending.TimestampUtc);
}
}
// Value dedup — but bypass it while recovering from Bad so a recovery whose value equals the
// pre-failure value still republishes Good (otherwise the node would stay Bad forever).
if (!st.LastPublishedBad && st.HasValue && Equals(st.LastValue, result.Value))
return;
if (failureReason is not null) PublishScriptLog(def, failureReason);
if (toPublish is { } p) Publish(def.RawPath, p.Value, p.StatusCode, pending.TimestampUtc);
}
/// <summary>Record a successful evaluation. MUST be called under <see cref="_lock"/>; returns what to
/// publish, or null when the value is deduped. The dedup is bypassed while recovering from Bad so a
/// recovery whose value equals the pre-failure value still republishes Good (otherwise the node would
/// stay Bad forever).</summary>
/// <param name="st">The tag's published state.</param>
/// <param name="value">The computed value.</param>
/// <param name="timestampUtc">The trigger's timestamp.</param>
/// <returns>The snapshot to publish, or null when deduped.</returns>
private (object? Value, uint StatusCode)? MarkEvaluationSucceeded(TagState st, object? value, DateTime timestampUtc)
{
if (!st.LastPublishedBad && st.HasValue && Equals(st.LastValue, value)) return null;
st.HasValue = true;
st.LastValue = result.Value;
st.LastValue = value;
st.LastPublishedBad = false;
st.LastTimestampUtc = timestampUtc;
_lastComputeUtc = DateTime.UtcNow;
Publish(rawPath, result.Value, StatusGood, timestampUtc);
return (value, StatusGood);
}
/// <summary>Failure tail (mirrors <c>VirtualTagActor</c> 02/S13): always emits the per-failure
/// script-log; additionally publishes a Bad snapshot — carrying the last-known value — once per
/// Good→Bad transition. The caller (<see cref="EvaluateTag"/>) only reaches here once every declared
/// dependency has arrived — the warm-up gate returns before the evaluator ever runs — so there is no
/// separate "inputs ready" condition to re-check here.</summary>
private void OnEvaluationFailed(CalculationTagDefinition def, TagState st, string reason, DateTime timestampUtc)
/// <summary>Failure tail (mirrors <c>VirtualTagActor</c> 02/S13): the caller always emits the
/// per-failure script-log; this additionally returns a Bad snapshot — carrying the last-known value —
/// once per Good→Bad transition, and null on a repeat failure. MUST be called under
/// <see cref="_lock"/>. Only reached once every declared dependency has arrived (the warm-up gate
/// returns before the evaluator ever runs), so there is no separate "inputs ready" condition here.</summary>
/// <param name="st">The tag's published state.</param>
/// <param name="timestampUtc">The trigger's timestamp.</param>
/// <returns>The Bad snapshot to publish on a fresh Good→Bad transition, else null.</returns>
private static (object? Value, uint StatusCode)? MarkEvaluationFailed(TagState st, DateTime timestampUtc)
{
PublishScriptLog(def, reason);
if (st.LastPublishedBad) return;
if (st.LastPublishedBad) return null;
st.LastPublishedBad = true;
st.LastTimestampUtc = timestampUtc;
// Carry the last-known value (null if none) with Bad quality.
Publish(def.RawPath, st.HasValue ? st.LastValue : null, StatusBadInternalError, timestampUtc);
return (st.HasValue ? st.LastValue : null, StatusBadInternalError);
}
private void Publish(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
{
var snapshot = new DataValueSnapshot(value, statusCode, timestampUtc, timestampUtc);
// Fired under _lock — the subscriber (DriverInstanceActor) marshals the event to its own thread,
// so there is no re-entrant call back into this driver.
// Fired outside both locks, from the single evaluation worker — so publish order still follows
// evaluation order, and a subscriber can neither deadlock against the state lock nor re-enter the
// driver behind it (DriverInstanceActor marshals the event onto its own thread regardless).
OnDataChange?.Invoke(this, new DataChangeEventArgs(SharedHandle, rawPath, snapshot));
}
@@ -453,7 +600,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
{
var timed = _tagsByRawPath.Values.Where(t => t.TimerInterval is not null).ToArray();
if (timed.Length == 0) return;
_timers = new CalculationTimerScheduler(EvaluateTagLocked, _logger);
_timers = new CalculationTimerScheduler(EvaluateTagFromTimer, _logger);
_timers.Start(timed);
}
@@ -466,6 +613,15 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
/// <summary>Test/diagnostic accessor: number of timer ticks skipped for in-flight groups.</summary>
internal long TimerSkippedTickCount => _timers?.SkippedTickCount ?? 0;
/// <summary>Test/diagnostic accessor: <see langword="true"/> when no change-triggered evaluation is
/// queued or in flight. A test that used to assert straight after <see cref="OnDependencyValue"/> —
/// which evaluated inline — waits on this instead, so the assertion still runs against a settled
/// driver rather than a sleep.</summary>
internal bool IsEvaluationIdle
{
get { lock (_lock) return _pending.Count == 0 && !_pumpScheduled; }
}
// A single shared handle for host-pushed publishes (change + timer). SubscribeAsync mints its own
// per-call handle for the initial-data callback; steady-state publishes reuse this so OnDataChange
// always carries a non-null handle.
@@ -478,6 +634,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
_disposed = true;
_running = false;
StopTimers();
lock (_lock) _pending.Clear();
_evaluator.Dispose();
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs
+94
View File
@@ -41,6 +41,18 @@ public sealed class CalculationDriverTests
return (driver, published, logs);
}
/// <summary>
/// Wait until the driver's evaluation pump has drained. A change-triggered evaluation is queued by
/// <c>OnDependencyValue</c> and run on the driver's own worker (Driver.Calculation-005), so a test
/// that used to assert on the very next line now settles first. Quiescence, not a sleep: the work
/// is enqueued synchronously inside the callback, so "idle" can never be observed before it lands.
/// </summary>
private static async Task Settle(CalculationDriver driver)
{
var idle = await WaitUntil(() => driver.IsEvaluationIdle, TimeSpan.FromSeconds(10));
idle.ShouldBeTrue("the calc driver's evaluation pump should have drained");
}
// ── dependency-ref extraction ─────────────────────────────────────────────────────────────
[Fact]
@@ -64,10 +76,12 @@ public sealed class CalculationDriverTests
// Only one of two deps arrived → the change-gate holds.
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.ShouldBeEmpty();
// Second dep arrives → now it computes + publishes 30.
driver.OnDependencyValue("src/b", 20.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var e).ShouldBeTrue();
e!.FullReference.ShouldBe("calc/e/sum");
@@ -90,12 +104,85 @@ public sealed class CalculationDriverTests
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.ShouldBeEmpty();
logs.ShouldBeEmpty("a still-missing declared dependency must hold the gate BEFORE evaluation, " +
"not merely before the publish");
}
// ── the dependency callback must not execute scripts (Driver.Calculation-005) ─────────────
/// <summary>
/// <c>OnDependencyValue</c> is called <b>directly and synchronously</b> from
/// <c>DependencyConsumerMuxAdapter.OnDependencyValueChanged</c> — i.e. on an Akka actor's own
/// dispatcher thread. Evaluating there means a script run (bounded only by <c>RunTimeout</c>,
/// default 2 s) occupies that thread, so the callback must queue the work and return instead.
/// <para>The publish is the observable: it can only happen after the script ran, so a publish on
/// the caller's own thread proves the script ran inline. The value assertion is the positive
/// control — without it this would pass just as happily against a driver that evaluates nothing.</para>
/// </summary>
[Fact]
public async Task A_dependency_arrival_does_not_run_the_script_on_the_callers_thread()
{
var (driver, published, _) = Build(
CalcTag("calc/e/dbl", "return (double)ctx.GetTag(\"src/a\").Value * 2.0;"));
var publishThreadId = 0;
driver.OnDataChange += (_, _) => publishThreadId = Environment.CurrentManagedThreadId;
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
var callerThreadId = Environment.CurrentManagedThreadId;
driver.OnDependencyValue("src/a", 5.0, StatusGood, DateTime.UtcNow);
(await WaitUntil(() => !published.IsEmpty, TimeSpan.FromSeconds(5)))
.ShouldBeTrue("the queued evaluation must still compute and publish the value");
published.TryDequeue(out var e).ShouldBeTrue();
e!.Snapshot.Value.ShouldBe(10.0);
publishThreadId.ShouldNotBe(callerThreadId,
"the calc script must run on the driver's own evaluation pump, never on the mux callback thread");
}
/// <summary>
/// The second half of Driver.Calculation-005: the evaluation also used to run <b>under the state
/// lock</b>, so a slow script froze every other lock-taking operation (here <c>ReadAsync</c>) for
/// its whole run as well as the callback itself. Both must now return while the script is still
/// executing; the trailing publish is the positive control that the script did run.
/// </summary>
[Fact]
public async Task A_slow_script_blocks_neither_the_dependency_callback_nor_a_concurrent_read()
{
// A CPU-bound busy-wait: the sandbox forbids Thread/Task, and DateTime is explicitly allowed.
const string slowScript =
"var deadline = System.DateTime.UtcNow.AddMilliseconds(600); " +
"while (System.DateTime.UtcNow < deadline) { } " +
"return (double)ctx.GetTag(\"src/a\").Value;";
var (driver, published, _) = Build(CalcTag("calc/e/slow", slowScript));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
var elapsed = System.Diagnostics.Stopwatch.StartNew();
driver.OnDependencyValue("src/a", 7.0, StatusGood, DateTime.UtcNow);
var callbackElapsed = elapsed.Elapsed;
// ReadAsync takes the same state lock the inline evaluation used to hold for the whole run.
var snapshot = await driver.ReadAsync(["calc/e/slow"], TestContext.Current.CancellationToken);
var readElapsed = elapsed.Elapsed;
callbackElapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(300),
"the mux callback thread must not wait on the script run");
readElapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(300),
"the state lock must not be held across the script run");
snapshot[0].StatusCode.ShouldBe(StatusBadWaitingForInitialData,
"the read got through while the evaluation was still in flight");
(await WaitUntil(() => !published.IsEmpty, TimeSpan.FromSeconds(10)))
.ShouldBeTrue("the slow evaluation must still complete and publish");
published.TryDequeue(out var e).ShouldBeTrue();
e!.Snapshot.Value.ShouldBe(7.0);
await driver.ShutdownAsync(TestContext.Current.CancellationToken);
}
// ── dedup + recompute ─────────────────────────────────────────────────────────────────────
[Fact]
@@ -105,9 +192,12 @@ public sealed class CalculationDriverTests
CalcTag("calc/e/dbl", "return (double)ctx.GetTag(\"src/a\").Value * 2.0;"));
await driver.InitializeAsync("{}", default);
// Each arrival is queued with the dependency snapshot that triggered it, so the three evaluations
// see 5 / 5 / 7 in that order however far the pump lags behind them.
driver.OnDependencyValue("src/a", 5.0, StatusGood, DateTime.UtcNow); // → 10 (publish)
driver.OnDependencyValue("src/a", 5.0, StatusGood, DateTime.UtcNow); // → 10 (dedup, no publish)
driver.OnDependencyValue("src/a", 7.0, StatusGood, DateTime.UtcNow); // → 14 (publish)
await Settle(driver);
published.Select(p => p.Snapshot.Value).ShouldBe(new object?[] { 10.0, 14.0 });
}
@@ -123,6 +213,7 @@ public sealed class CalculationDriverTests
// a = 0 → DivideByZero → Failure → Bad (carrying last-known = null) + one script-log.
driver.OnDependencyValue("src/a", 0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var bad).ShouldBeTrue();
bad!.Snapshot.StatusCode.ShouldBe(StatusBadInternalError);
@@ -133,11 +224,13 @@ public sealed class CalculationDriverTests
// a = 0 again → still failing, but NOT a fresh transition → no second Bad publish (script-log still emits).
driver.OnDependencyValue("src/a", 0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.ShouldBeEmpty();
logs.Count.ShouldBe(2);
// a = 5 → 20 → Good recovery is force-published.
driver.OnDependencyValue("src/a", 5, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var good).ShouldBeTrue();
good!.Snapshot.Value.ShouldBe(20);
@@ -175,6 +268,7 @@ public sealed class CalculationDriverTests
before[0].StatusCode.ShouldBe(StatusBadWaitingForInitialData);
driver.OnDependencyValue("src/a", 3.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
var after = await driver.ReadAsync(new[] { "calc/e/x" }, default);
after[0].StatusCode.ShouldBe(StatusGood);
after[0].Value.ShouldBe(6.0);