diff --git a/code-reviews/Driver.Calculation.Contracts/findings.md b/code-reviews/Driver.Calculation.Contracts/findings.md index c17b915b..f5d14409 100644 --- a/code-reviews/Driver.Calculation.Contracts/findings.md +++ b/code-reviews/Driver.Calculation.Contracts/findings.md @@ -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. diff --git a/code-reviews/Driver.Calculation/findings.md b/code-reviews/Driver.Calculation/findings.md index 1b4c31a5..0f9ec50c 100644 --- a/code-reviews/Driver.Calculation/findings.md +++ b/code-reviews/Driver.Calculation/findings.md @@ -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. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs index 37ca4929..1e6954f0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Contracts/CalculationTagDefinitionFactory.cs @@ -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 } } + /// + /// The field-by-field read, over an already-parsed root. Split out of + /// so — 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 scriptId 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. + /// + /// The parsed TagConfig root element. + /// The read fields when this returns . + /// when the root is an object naming a script. + 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; + } + /// /// Deploy-time inspection of a calc tag's TagConfig blob. Warns for a blob /// 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 " 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..35945878 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,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) ───────────── + + /// + /// 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 that + /// lands on the caller's stack between 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 + /// ids 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. + /// + [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(); + 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"); + } + + /// + /// 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 +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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationTagDefinitionFactoryInspectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationTagDefinitionFactoryInspectTests.cs new file mode 100644 index 00000000..fac0c345 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationTagDefinitionFactoryInspectTests.cs @@ -0,0 +1,111 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests; + +/// +/// Deploy-time gate contract for . 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 — Inspect now reads its own already-parsed root instead of +/// re-parsing the blob through TryReadTagConfig) is provably behaviour-preserving. +/// The invariant underneath all of it: Inspect is never STRICTER than +/// — the deploy gate can be configured +/// to turn warnings into blocking errors, so it may never flag a blob the driver serves happily. +/// +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.", + ]); + } + + /// The rejection and the advisories are independent: a blob can draw both, and the rejection + /// always comes first. + [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'"); + } + + /// 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. + [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(); + } +}