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

merge: Calculation review findings resolved (Driver.Calculation-005 pump split, Driver.Calculation.Contracts-001 single-parse Inspect)

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-10 08:09:21 -04:00
parent d7568ac342 5c8d6f6f98
commit 55f4354117
6 changed files with 547 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-reviews/Driver.Calculation.Contracts/findings.md
+23 -3
View File
@@ -7,7 +7,7 @@
| Review date | 2026-08-10 |
| Commit reviewed | `b990adae` |
| Status | Reviewed |
| Open findings | 1 |
| Open findings | 0 |
## Checklist coverage
@@ -33,7 +33,7 @@
| Severity | Low |
| Category | Performance & resource management |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs:107-160` (`Inspect`) |
| Status | Open |
| Status | Resolved |
**Description:** `Inspect` parses `reference` with `JsonDocument.Parse` once directly (to check the root
kind and read `changeTriggered`/`timerIntervalMs` for the "present-but-discarded" warnings), and then —
@@ -52,4 +52,24 @@ and `AdminUI`'s `CalculationTagConfigModel.cs`), so widening it safely is a smal
a same-file one-liner, and the current cost is one extra JSON parse per calc tag at deploy time — not a
runtime cost.
**Resolution:** _(Open — Low severity, record-only per review policy; deploy-time only, not a hot path.)_
**Resolution (2026-08-10):** Fixed in `73230a42` — and it turned out to be a same-file change after all,
because widening the *public* signature was never necessary. The field-by-field read moved into a private
`TryReadRoot(JsonElement root, out CalculationTagConfigFields fields)`; `TryReadTagConfig(string, …)` keeps
its exact signature (still the entry point `Driver.Calculation`'s `CalculationTagDefinition.TryFromTagConfig`
and AdminUI's `CalculationTagConfigModel` consume) and now does nothing but parse, guard `JsonException`, and
delegate. `Inspect` calls `TryReadRoot(root, …)` over the root it has already parsed, so the blob is parsed
exactly once per gate pass while the scriptId rule stays owned by one authority — the "one authority" design
the finding names is preserved, not traded away.
Equivalence is exact rather than argued: by the time `Inspect` reaches that call it has already returned early
for a blank blob, a non-`{`-prefixed blob, and a `JsonException`, and has already established
`root.ValueKind == Object` — precisely the three pre-checks `TryReadTagConfig` performs before its own
field read, over the same default-options parse of the same string.
Pinned by a new `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationTagDefinitionFactoryInspectTests.cs`
(17 cases: exact warning wording, the rejection-vs-advisory split and its ordering, the blank/non-object
pre-filter, and a falsification control that every advisory-drawing blob is still ACCEPTED by
`TryReadTagConfig` — the "never stricter than the runtime read" invariant). `Inspect` previously had no
coverage outside the DB-backed AdminUI parity corpus, so these were written against the **pre-refactor**
code and verified to pass there (production change stashed) before the refactor landed — a pin, not a
description of the new code. Behaviour is byte-identical; no accept/reject/advisory outcome changed.
code-reviews/Driver.Calculation/findings.md
+64 -4
View File
@@ -7,7 +7,7 @@
| Review date | 2026-08-10 |
| Commit reviewed | `b990adae` |
| Status | Reviewed |
| Open findings | 1 |
| Open findings | 0 |
## Checklist coverage
@@ -15,7 +15,7 @@
|---|---|---|
| 1 | Correctness & logic bugs | 3 findings (Driver.Calculation-001, -002, -003), all Resolved |
| 2 | OtOpcUa conventions | No issues found. Galaxy/MXAccess-specific conventions do not apply to this pseudo-driver (no transport, no gateway). It correctly implements the capability-interface seams (IDriver, IDependencyConsumer, ISubscribable, IReadable, ISkippedTagReporter), .NET 10/AnyCPU, and authorization stays out of driver-level code. |
| 3 | Concurrency & thread safety | 1 finding (Driver.Calculation-005), Open (cross-module) |
| 3 | Concurrency & thread safety | 1 finding (Driver.Calculation-005), Resolved |
| 4 | Error handling & resilience | Covered by Driver.Calculation-002 (script-evaluation faults never throw out of the driver; SkippedTagLedger tolerates a single bad tag without failing the whole rebuild). No further issues found. |
| 5 | Security | No issues found. No secrets, no I/O, no SQL; the sandboxed Roslyn evaluator (shared Core.Scripting sandbox) is the security boundary and lives outside this module. |
| 6 | Performance & resource management | Covered by Driver.Calculation-005 (blocking evaluation under a shared lock). CalculationTimerScheduler correctly bounds outstanding work to one tick per interval group (skip-if-in-flight) and CalculationEvaluator.ClearCompiledScripts() is invoked on every ReinitializeAsync, so compiled-script AssemblyLoadContexts do not accrete across script edits. No further issues found. |
@@ -186,7 +186,7 @@ test — noted above, not silently dropped.
| Severity | Medium |
| Category | Concurrency & thread safety |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs:300-314` (`OnDependencyValue`), `:374-410` (`EvaluateTag`); cross-module: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs` |
| Status | Open |
| Status | Resolved |
**Description:** `OnDependencyValue` evaluates every change-triggered dependent tag **inline, under
`_lock`**, and `EvaluateTag`'s call into `CalculationEvaluator.Evaluate` ultimately runs
@@ -222,4 +222,64 @@ asynchronously) so the actor thread returns immediately; or (c) tightening `RunT
independent of the `RunTimeoutMs` config knob. Any of these should be scoped and reviewed together with the
equivalent `VirtualTagActor` behaviour so the two mirrors do not diverge.
**Resolution:** _(Open — cross-module, not fixed as part of this review.)_
**Resolution (2026-08-10):** Fixed in `235e06a8` — **inside this module, with no change to the
`IDependencyConsumer` contract and none to `Runtime`.** The finding scoped a fix as necessarily
cross-module (option (a), an async `OnDependencyValue`), but option (b) — the driver owning its own
dispatch — closes both halves without touching a shared interface, which is the smaller change and leaves
the mirrored `VirtualTagActor` behaviour free to be addressed separately.
`OnDependencyValue` no longer evaluates anything. Under `_lock` it records the arrival and *prepares* each
change-triggered dependent — resolve the definition, apply the warm-up gate, snapshot that tag's declared
dependency values — into a `PendingEvaluation`, enqueues them, and returns; a single `Task.Run` **evaluation
pump** drains the queue in FIFO order. `_lock` is a short-held state lock again (dictionary work only, never
held across a script run); a new `_evalGate` carries the mutual exclusion `_lock` used to provide, and is
taken **only** by the two evaluation workers (the pump and `EvaluateTagFromTimer`), never by
`OnDependencyValue`, `ReadAsync` or `ReinitializeAsync` — which is what makes "no host thread can block on a
script" structural rather than incidental. Lock order is always `_evalGate` → `_lock`. `EvaluateTag` split
into `PrepareEvaluation` (under `_lock`) / `RunEvaluation` (script + apply, outside it) /
`MarkEvaluationSucceeded` + `MarkEvaluationFailed` (state decisions, under `_lock`, returning what to
publish); the script log and the `OnDataChange` fan-out now fire outside both locks.
**The guarantee the fix rests on: at most one evaluation runs at a time.** The pump is single (`_pumpScheduled`
is set and cleared under `_lock`, and the flag and the queue are always checked together, so an enqueue either
starts a pump or is seen by one that has not yet exited), and `_evalGate` serializes it against timer ticks —
so per-tag *and* cross-tag evaluation ordering is exactly what it was when the change path evaluated inline
under the state lock. Publishes follow evaluation order because they are issued by the same serialized worker.
Semantics deliberately preserved beyond ordering: the warm-up gate still runs BEFORE the evaluator
(Driver.Calculation-002 is not regressed — it moved into `PrepareEvaluation`, which is equivalent because
`_arrivedDeps` only ever grows and the arrival that completes the set queues its own evaluation); dedup,
once-per-transition Bad, and force-published recovery are unchanged. **The dependency snapshot is taken at
trigger time, not when the pump reaches the item** — without that, a burst of arrivals would collapse onto
whichever value happened to be current when the pump caught up, changing which intermediate values publish
(`Equal_result_is_deduped_and_a_changed_result_republishes` is the case that would have silently become
`[14.0]` instead of `[10.0, 14.0]`). The timer path keeps its own structure — a tick still evaluates on the
scheduler's pool thread under the scheduler's per-group in-flight guard, which is what bounds outstanding
timer work; routing ticks through the pump instead would have made `SkippedTickCount` dead and let ticks
queue without bound. `ReinitializeAsync`, `ShutdownAsync` and `Dispose` clear the queue (prepared items are
stale against a rebuilt tag table), and `RunEvaluation` re-looks-up the tag's state under `_lock` after the
run so a tag dropped mid-evaluation discards its result rather than publishing it.
RED first, two tests, both failing against the pre-fix code:
`A_dependency_arrival_does_not_run_the_script_on_the_callers_thread` (the publish — which can only happen
after the script ran — landed on the caller's own stack, inside the callback) and
`A_slow_script_blocks_neither_the_dependency_callback_nor_a_concurrent_read` (a 600 ms busy-wait script:
the callback took 1.21 s, and `ReadAsync`, which needs the same state lock, was stuck behind it). Both
carry a positive control asserting the value still computes and publishes, so neither can pass against a
driver that simply evaluates nothing. ⚠️ The first test's original form compared the publish's **thread id**
against the caller's and was itself flaky (1 failure in 6 runs): the pump is free to reuse the pooled thread
the test released while awaiting, so an id match means nothing. It asserts on a thread-local flag set for
exactly the callback window instead — which is exact, and was re-falsified by making `SchedulePump` run the
pump synchronously (red immediately, then reverted).
Known trade, documented rather than hidden: the pending queue is unbounded. Under sustained arrival
pressure the previous inline path applied backpressure by blocking the mux actor — which is the defect —
so the queue is the relief valve, and the tag-set is small and bounded by authored config. Five existing
tests that asserted on the line after `OnDependencyValue` now settle on a new internal `IsEvaluationIdle`
probe first (quiescence, not a sleep — the work is enqueued synchronously inside the callback, so idle
cannot be observed before it lands). Suites green: `Driver.Calculation.Tests` 53/53 (34 before this work),
`Runtime.Tests` 558 passed / 3 skipped.
The mirrored `VirtualTagActor.OnDependencyChanged` blocking shape this finding also names is **unchanged** —
it lives in `Runtime`, outside this module, and is left as the separate scoped change the finding's
recommendation asks for.
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs
+39 -21
View File
@@ -68,26 +68,7 @@ public static class CalculationTagDefinitionFactory
try
{
using var doc = JsonDocument.Parse(tagConfig);
var root = doc.RootElement;
if (root.ValueKind != JsonValueKind.Object) return false;
var scriptId = root.TryGetProperty("scriptId", out var sidEl) && sidEl.ValueKind == JsonValueKind.String
? sidEl.GetString() : null;
if (string.IsNullOrWhiteSpace(scriptId)) return false;
var scriptSource = root.TryGetProperty("scriptSource", out var srcEl) && srcEl.ValueKind == JsonValueKind.String
? srcEl.GetString() ?? string.Empty : string.Empty;
// changeTriggered defaults to true (mini-design §2) — only an explicit false disables it.
var changeTriggered = !(root.TryGetProperty("changeTriggered", out var ctEl)
&& ctEl.ValueKind == JsonValueKind.False);
var timerMs = root.TryGetProperty("timerIntervalMs", out var tEl)
&& tEl.ValueKind == JsonValueKind.Number && tEl.TryGetInt32(out var ms) && ms > 0
? ms : (int?)null;
fields = new CalculationTagConfigFields(scriptId!, scriptSource, changeTriggered, timerMs);
return true;
return TryReadRoot(doc.RootElement, out fields);
}
catch (JsonException)
{
@@ -95,6 +76,41 @@ public static class CalculationTagDefinitionFactory
}
}
/// <summary>
/// The field-by-field read, over an already-parsed root. Split out of
/// <see cref="TryReadTagConfig"/> so <see cref="Inspect"/> — which has necessarily parsed the blob
/// already to reach its root-kind and present-but-discarded checks — can consult the one authority
/// on what a usable <c>scriptId</c> is without parsing the same string a second time
/// (Driver.Calculation.Contracts-001). The public string-based entry point is what other modules
/// consume, so it keeps its signature.
/// </summary>
/// <param name="root">The parsed <c>TagConfig</c> root element.</param>
/// <param name="fields">The read fields when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when the root is an object naming a script.</returns>
private static bool TryReadRoot(JsonElement root, out CalculationTagConfigFields fields)
{
fields = default;
if (root.ValueKind != JsonValueKind.Object) return false;
var scriptId = root.TryGetProperty("scriptId", out var sidEl) && sidEl.ValueKind == JsonValueKind.String
? sidEl.GetString() : null;
if (string.IsNullOrWhiteSpace(scriptId)) return false;
var scriptSource = root.TryGetProperty("scriptSource", out var srcEl) && srcEl.ValueKind == JsonValueKind.String
? srcEl.GetString() ?? string.Empty : string.Empty;
// changeTriggered defaults to true (mini-design §2) — only an explicit false disables it.
var changeTriggered = !(root.TryGetProperty("changeTriggered", out var ctEl)
&& ctEl.ValueKind == JsonValueKind.False);
var timerMs = root.TryGetProperty("timerIntervalMs", out var tEl)
&& tEl.ValueKind == JsonValueKind.Number && tEl.TryGetInt32(out var ms) && ms > 0
? ms : (int?)null;
fields = new CalculationTagConfigFields(scriptId!, scriptSource, changeTriggered, timerMs);
return true;
}
/// <summary>
/// Deploy-time inspection of a calc tag's <c>TagConfig</c> blob. Warns for a blob
/// <see cref="TryReadTagConfig"/> rejects (unparseable JSON, a non-object root, or no usable
@@ -130,7 +146,9 @@ public static class CalculationTagDefinitionFactory
return warnings;
}
if (!TryReadTagConfig(reference, out _))
// The SAME read the runtime performs, over the root already parsed above — never a second
// parse of the same string, and never a reimplementation of the scriptId rule.
if (!TryReadRoot(root, out _))
{
warnings.Add(
"Calculation TagConfig has no usable 'scriptId' (a non-blank string) — the driver does not "
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
+99
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,90 @@ 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 that
/// lands on the caller's stack <em>between</em> entering and leaving the callback proves the script
/// ran inline. A thread-local flag set for exactly that window is the exact test — comparing thread
/// <i>ids</i> is not, because the pump is free to reuse the pooled thread the test released while
/// waiting, which makes an id match mean nothing. The value assertion is the positive control:
/// without it this would pass just as happily against a driver that evaluates nothing at all.</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;"));
using var insideCallback = new ThreadLocal<bool>();
var publishedInline = false;
driver.OnDataChange += (_, _) => { if (insideCallback.Value) publishedInline = true; };
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
insideCallback.Value = true;
driver.OnDependencyValue("src/a", 5.0, StatusGood, DateTime.UtcNow);
insideCallback.Value = false;
(await WaitUntil(() => !published.IsEmpty, TimeSpan.FromSeconds(10)))
.ShouldBeTrue("the queued evaluation must still compute and publish the value");
published.TryDequeue(out var e).ShouldBeTrue();
e!.Snapshot.Value.ShouldBe(10.0);
publishedInline.ShouldBeFalse(
"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 +197,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 +218,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 +229,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 +273,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);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationTagDefinitionFactoryInspectTests.cs
+111
View File
@@ -0,0 +1,111 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
/// <summary>
/// Deploy-time gate contract for <see cref="CalculationTagDefinitionFactory.Inspect"/>. Its outcomes
/// were pinned only by the AdminUI parity corpus (a DB-backed suite); these fix the wording and the
/// rejection/advisory split locally, so the single-parse refactor
/// (Driver.Calculation.Contracts-001 — <c>Inspect</c> now reads its own already-parsed root instead of
/// re-parsing the blob through <c>TryReadTagConfig</c>) is provably behaviour-preserving.
/// <para>The invariant underneath all of it: <c>Inspect</c> is never STRICTER than
/// <see cref="CalculationTagDefinitionFactory.TryReadTagConfig"/> — the deploy gate can be configured
/// to turn warnings into blocking errors, so it may never flag a blob the driver serves happily.</para>
/// </summary>
public sealed class CalculationTagDefinitionFactoryInspectTests
{
[Fact]
public void A_clean_blob_draws_nothing()
{
CalculationTagDefinitionFactory
.Inspect("""{"scriptId":"s1","scriptSource":"return 1.0;","changeTriggered":true,"timerIntervalMs":5000}""")
.ShouldBeEmpty();
}
[Theory]
[InlineData("")]
[InlineData(" ")]
[InlineData("[1,2]")] // not object-shaped at all — some other authoring surface's blob, not ours
[InlineData("plain text")]
public void A_blank_or_non_object_shaped_blob_is_left_alone(string blob)
{
CalculationTagDefinitionFactory.Inspect(blob).ShouldBeEmpty();
}
[Fact]
public void Unparseable_json_is_reported_as_a_skip()
{
CalculationTagDefinitionFactory.Inspect("{\"scriptId\":").ShouldBe(
[
"Calculation TagConfig is not valid JSON — the driver will skip this tag and it will never compute.",
]);
}
// The "root is not a JSON object" warning is unreachable from Inspect by construction: the leading-'{'
// pre-filter above returns early for every non-object root, and anything that starts with '{' and
// parses IS an object. It is kept as a belt-and-braces guard for a future change to that pre-filter,
// so there is nothing to pin here beyond the pre-filter cases already covered.
[Theory]
[InlineData("{}")]
[InlineData("""{"scriptId":" "}""")]
[InlineData("""{"scriptId":42}""")]
public void A_blob_with_no_usable_scriptId_is_reported_as_a_skip(string blob)
{
CalculationTagDefinitionFactory.Inspect(blob).ShouldContain(
"Calculation TagConfig has no usable 'scriptId' (a non-blank string) — the driver does not "
+ "treat this as a calc tag, skips it, and it never computes.");
}
[Fact]
public void A_non_boolean_changeTriggered_draws_the_present_but_discarded_advisory()
{
CalculationTagDefinitionFactory.Inspect("""{"scriptId":"s1","changeTriggered":"false"}""").ShouldBe(
[
"value '\"false\"' for 'changeTriggered' is not a JSON boolean; anything but an explicit false "
+ "is read as TRUE, so this tag is change-triggered whatever the value looks like.",
]);
}
[Theory]
[InlineData("0")]
[InlineData("-5")]
[InlineData("1.5")]
[InlineData("\"5000\"")]
public void A_timerIntervalMs_that_is_not_a_positive_whole_number_draws_the_present_but_discarded_advisory(string raw)
{
var warnings = CalculationTagDefinitionFactory.Inspect($$"""{"scriptId":"s1","timerIntervalMs":{{raw}}}""");
warnings.ShouldBe(
[
$"value '{raw}' for 'timerIntervalMs' is not a positive whole number of milliseconds; it is "
+ "discarded and the tag gets NO periodic re-evaluation.",
]);
}
/// <summary>The rejection and the advisories are independent: a blob can draw both, and the rejection
/// always comes first.</summary>
[Fact]
public void A_rejected_blob_still_reports_its_discarded_fields()
{
var warnings = CalculationTagDefinitionFactory.Inspect("""{"changeTriggered":1,"timerIntervalMs":0}""");
warnings.Count.ShouldBe(3);
warnings[0].ShouldStartWith("Calculation TagConfig has no usable 'scriptId'");
warnings[1].ShouldStartWith("value '1' for 'changeTriggered'");
warnings[2].ShouldStartWith("value '0' for 'timerIntervalMs'");
}
/// <summary>Falsification control for the "never stricter than the runtime read" invariant: every blob
/// the runtime ACCEPTS but this gate warns about must still be accepted by the runtime read, so an
/// advisory can never be mistaken for a rejection.</summary>
[Theory]
[InlineData("""{"scriptId":"s1","changeTriggered":"yes"}""")]
[InlineData("""{"scriptId":"s1","timerIntervalMs":0}""")]
public void An_advisory_never_means_the_runtime_read_rejects_the_blob(string blob)
{
CalculationTagDefinitionFactory.Inspect(blob).ShouldNotBeEmpty();
CalculationTagDefinitionFactory.TryReadTagConfig(blob, out _).ShouldBeTrue();
}
}