diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
index 4e5c2065..2363b539 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
@@ -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).
///
///
+/// Where scripts run (Driver.Calculation-005). is called
+/// directly and synchronously by DependencyConsumerMuxAdapter — on an Akka actor's own
+/// dispatcher thread — so it does no evaluation itself: it records the value, prepares the
+/// affected tags (definition + dependency snapshot, taken at trigger time) and hands them to a
+/// single serialized evaluation pump. Script execution therefore never runs on the mux
+/// callback thread and never runs under ; the pump and the timer ticks are
+/// serialized against each other by instead, which preserves the
+/// one-evaluation-at-a-time ordering the state lock used to provide.
+///
+///
/// Error semantics (mini-design §6). An evaluator Failure publishes Bad quality
/// carrying the last-known value, once per Good→Bad transition and only after all deps have arrived,
/// and emits a ScriptLogEntry on the script-logs 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 _dependencyRefs = Array.Empty();
- // 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 _depValues = new(StringComparer.Ordinal);
private readonly HashSet _arrivedDeps = new(StringComparer.Ordinal);
private readonly Dictionary _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 _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? 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 ----
- /// 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).
- private void EvaluateTagLocked(string rawPath)
+ /// One prepared evaluation: the calc tag's definition, the dependency snapshot taken when its
+ /// trigger fired, and that trigger's timestamp. Built under by
+ /// and executed outside it, so the script itself needs no shared state.
+ /// The calc tag being evaluated.
+ /// Its declared reads' values, as of the trigger.
+ /// The trigger's timestamp, stamped on whatever is published.
+ private sealed record PendingEvaluation(
+ CalculationTagDefinition Definition,
+ IReadOnlyDictionary Dependencies,
+ DateTime TimestampUtc);
+
+ ///
+ /// 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 . Returns
+ /// when the tag must not evaluate — unknown, no resolved script source
+ /// (logged at init), or still warming up.
+ /// 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
+ /// 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. Applying it here rather than on the pump is equivalent:
+ /// only ever grows, and the arrival that completes the set queues its own
+ /// evaluation.
+ /// 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.
+ ///
+ /// The calc tag to prepare.
+ /// The trigger's timestamp.
+ /// The prepared evaluation, or null when the tag must not evaluate.
+ 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
- /// Evaluate one calc tag and publish the outcome. MUST be called under .
- 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(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);
+ /// Start the evaluation pump if it is not already running. MUST be called under
+ /// — 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.
+ private void SchedulePump()
+ {
+ if (_pumpScheduled) return;
+ _pumpScheduled = true;
+ _ = Task.Run(RunEvaluationPump);
+ }
- if (!result.Success)
+ /// 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.
+ 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);
+ }
+ }
+ }
+ }
+
+ /// 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).
+ 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);
+ }
+
+ /// Run one prepared evaluation and publish the outcome. MUST be called under
+ /// and NOT under : the script run is the whole point of the
+ /// split, and the script log + fan-out both invoke subscriber code.
+ /// The prepared evaluation.
+ 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);
+ }
+
+ /// Record a successful evaluation. MUST be called under ; 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).
+ /// The tag's published state.
+ /// The computed value.
+ /// The trigger's timestamp.
+ /// The snapshot to publish, or null when deduped.
+ 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);
}
- /// Failure tail (mirrors VirtualTagActor 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 () 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.
- private void OnEvaluationFailed(CalculationTagDefinition def, TagState st, string reason, DateTime timestampUtc)
+ /// Failure tail (mirrors VirtualTagActor 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
+ /// . 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.
+ /// The tag's published state.
+ /// The trigger's timestamp.
+ /// The Bad snapshot to publish on a fresh Good→Bad transition, else null.
+ 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
/// Test/diagnostic accessor: number of timer ticks skipped for in-flight groups.
internal long TimerSkippedTickCount => _timers?.SkippedTickCount ?? 0;
+ /// Test/diagnostic accessor: when no change-triggered evaluation is
+ /// queued or in flight. A test that used to assert straight after —
+ /// which evaluated inline — waits on this instead, so the assertion still runs against a settled
+ /// driver rather than a sleep.
+ 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();
}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs
index 276b7f13..5f9a9160 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs
@@ -41,6 +41,18 @@ public sealed class CalculationDriverTests
return (driver, published, logs);
}
+ ///
+ /// Wait until the driver's evaluation pump has drained. A change-triggered evaluation is queued by
+ /// OnDependencyValue 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.
+ ///
+ 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) ─────────────
+
+ ///
+ /// OnDependencyValue is called directly and synchronously from
+ /// DependencyConsumerMuxAdapter.OnDependencyValueChanged — i.e. on an Akka actor's own
+ /// dispatcher thread. Evaluating there means a script run (bounded only by RunTimeout,
+ /// default 2 s) occupies that thread, so the callback must queue the work and return instead.
+ /// 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.
+ ///
+ [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");
+ }
+
+ ///
+ /// The second half of Driver.Calculation-005: the evaluation also used to run under the state
+ /// lock, so a slow script froze every other lock-taking operation (here ReadAsync) 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.
+ ///
+ [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);