using System.Text.Json;
using System.Text.Json.Serialization;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
///
/// The Calculation pseudo-driver. Its tags are ordinary raw tags whose value is computed
/// by a C# script over other tags' live values — there is no backend, no endpoint, no discovery, no
/// writes. It implements:
///
/// - — lifecycle + always-Connected health.
/// - — the host feeds it the other tags' live values.
/// - — computed values flow out via
/// exactly like any driver, so the ordinary node fan-out + the mux re-entry (calc-of-calc) apply.
/// - — on-demand read returns the last computed snapshot
/// (BadWaitingForInitialData before the first evaluation).
///
///
///
/// Triggers. A tag re-evaluates on any of its declared dependencies changing (change-trigger,
/// gated exactly like VirtualTagActor: nothing publishes until every declared dep has arrived
/// 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
/// 's ScriptLogTopicSink). Recovery to Good is force-published.
/// A historized Bad records BadInternalError.
///
///
public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscribable, IReadable, ISkippedTagReporter, IDisposable
{
/// Authored tags this driver could not build into its calc table (Gitea #541). A raw tag with
/// no scriptId is not a calc tag and computes nothing.
private readonly SkippedTagLedger _skippedTags = new();
///
public SkippedTagReport? GetSkippedTags() => _skippedTags.Snapshot();
// OPC UA status codes surfaced by the calc driver.
private const uint StatusGood = 0u;
private const uint StatusBadWaitingForInitialData = 0x80320000u;
private const uint StatusBadInternalError = 0x80020000u;
private readonly string _driverInstanceId;
private readonly CalculationDriverOptions _options;
private readonly ILogger _logger;
private readonly ScriptRootLogger _scriptRoot;
private readonly CalculationEvaluator _evaluator;
// Authored calc tags, keyed by RawPath (identity + evaluation id). Built at construction so
// DependencyRefs is available BEFORE InitializeAsync (the host reads it when spawning the mux-adapter),
// and REBUILT on every ReinitializeAsync (an in-place ApplyDelta re-binds an edited calc-tag set).
private readonly Dictionary _tagsByRawPath = new(StringComparer.Ordinal);
// rawPath (a dependency) → the calc tags that read it AND are change-triggered. Drives OnDependencyValue.
private readonly Dictionary> _changeDependents = new(StringComparer.Ordinal);
// Union of every tag's declared dependency refs — the interest set the host registers on the mux.
// 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 (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();
// Dependency SNAPSHOTS, not bare values (archreview 05/STAB-19): the status code is what tells a
// comms-lost input apart from a genuine Good null, and storing values only is what let this driver
// compute a Good output from a dead source.
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;
private bool _disposed;
/// Occurs when a calc tag's computed value changes (or transitions Good↔Bad).
public event EventHandler? OnDataChange;
/// Initializes a new .
/// Bound options carrying the authored raw calc tags.
/// Stable logical id of this driver instance.
/// Root script logger — user ctx.Logger output + failure entries fan out
/// to the script-logs topic through its ScriptLogTopicSink. When null a no-op logger is used.
/// Host-side diagnostics logger; defaults to the null logger.
public CalculationDriver(
CalculationDriverOptions options,
string driverInstanceId,
ScriptRootLogger? scriptRoot = null,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
_options = options;
_driverInstanceId = driverInstanceId;
_logger = logger ?? NullLogger.Instance;
_scriptRoot = scriptRoot ?? new ScriptRootLogger(new LoggerConfiguration().CreateLogger());
// _logger (ILogger) implements the non-generic ILogger, so it is passed straight
// through — the evaluator's host-side diagnostics (compile errors, sandbox violations, execution
// throws) land on the SAME logger the driver itself was constructed with, not a separate one.
_evaluator = new CalculationEvaluator(_logger, _scriptRoot, _options.RunTimeout);
RebuildTagTable(_options);
}
///
/// (Re)build the authored tag table (), the change-trigger fan-out
/// (), the dependency-ref interest set (),
/// and the per-tag state map () from . Called
/// once at construction AND on every so an in-place ApplyDelta — a
/// calc-tag add/remove/edit, or a script-source edit — actually takes effect. Extracted so the ctor
/// and reinit paths share ONE construction routine and cannot drift.
/// Per-tag is preserved for a tag that survives the edit (cheap last-known
/// value carry-over); a brand-new tag starts fresh; a removed tag's state is dropped. MUST be called
/// under when invoked post-construction (the ctor runs single-threaded).
///
/// The (new) bound options carrying the authored raw calc tags.
private void RebuildTagTable(CalculationDriverOptions config)
{
_tagsByRawPath.Clear();
_changeDependents.Clear();
// Retain surviving tags' state; anything not re-authored is dropped when we swap this in below.
var retainedState = new Dictionary(StringComparer.Ordinal);
_skippedTags.BeginRebuild(config.RawTags.Count);
foreach (var entry in config.RawTags)
{
if (!CalculationTagDefinition.TryFromTagConfig(entry.TagConfig, entry.RawPath, out var def))
{
_skippedTags.Skip(entry.RawPath, "no scriptId — not a calc tag");
_logger.LogWarning(
"Calculation {Driver}: raw tag {RawPath} has no scriptId; skipping (not a calc tag)",
_driverInstanceId, entry.RawPath);
continue;
}
if (string.IsNullOrWhiteSpace(def.ScriptSource))
{
_logger.LogWarning(
"Calculation {Driver}: calc tag {RawPath} (script {Script}) has no resolved scriptSource; " +
"it will not compute until the deploy artifact injects the script source",
_driverInstanceId, entry.RawPath, def.ScriptId);
}
_tagsByRawPath[entry.RawPath] = def;
// Carry the last-known state forward for a surviving tag; a brand-new tag starts fresh.
retainedState[entry.RawPath] =
_stateByRawPath.TryGetValue(entry.RawPath, out var prior) ? prior : new TagState();
if (def.ChangeTriggered)
{
foreach (var dep in def.DependencyRefs)
{
if (!_changeDependents.TryGetValue(dep, out var list))
_changeDependents[dep] = list = new List();
if (!list.Contains(entry.RawPath, StringComparer.Ordinal)) list.Add(entry.RawPath);
}
}
}
_skippedTags.Commit();
// Swap in the retained/new state map — a removed tag's state is dropped so it stops publishing.
_stateByRawPath.Clear();
foreach (var (rawPath, state) in retainedState) _stateByRawPath[rawPath] = state;
_dependencyRefs = _tagsByRawPath.Values
.SelectMany(t => t.DependencyRefs)
.Distinct(StringComparer.Ordinal)
.ToArray();
}
// Reinit config-bind options — mirrors the factory's bind (case-insensitive, enum-string, trailing-comma
// tolerant) so an ApplyDelta payload re-binds identically to the spawn payload.
private static readonly JsonSerializerOptions ReinitJsonOptions = new()
{
PropertyNameCaseInsensitive = true,
ReadCommentHandling = JsonCommentHandling.Skip,
AllowTrailingCommas = true,
Converters = { new JsonStringEnumConverter() },
};
/// Re-bind the authored raw calc tags from a new merged DriverConfig JSON (the ApplyDelta
/// payload). The ctor-fixed timeout is kept — RunTimeout is bound
/// at spawn (a change to it is out of this in-place path's scope). A parse failure logs and keeps the
/// current tags so a malformed delta can never blank the driver.
private CalculationDriverOptions ParseOptions(string driverConfigJson)
{
if (!string.IsNullOrWhiteSpace(driverConfigJson))
{
try
{
var dto = JsonSerializer.Deserialize(driverConfigJson, ReinitJsonOptions);
if (dto is not null)
{
return new CalculationDriverOptions
{
RawTags = dto.RawTags is { Count: > 0 } ? [.. dto.RawTags] : [],
RunTimeout = _options.RunTimeout,
};
}
}
catch (JsonException ex)
{
_logger.LogWarning(ex,
"Calculation {Driver}: reinit config parse failed; keeping the current tag table",
_driverInstanceId);
}
}
return new CalculationDriverOptions { RawTags = _options.RawTags, RunTimeout = _options.RunTimeout };
}
/// Minimal reinit-bind DTO — the calc driver has no backend config, only the injected raw tags
/// (+ an optional eval-timeout, ignored on reinit since the evaluator is ctor-fixed). Mirrors the factory's
/// CalculationDriverConfigDto.
private sealed class ReinitConfigDto
{
public List? RawTags { get; init; }
public int? RunTimeoutMs { get; init; }
}
// ---- IDriver ----
///
public string DriverInstanceId => _driverInstanceId;
///
public string DriverType => DriverTypeNames.Calculation;
///
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
_running = true;
StartTimers();
_logger.LogInformation(
"Calculation {Driver}: initialized with {TagCount} calc tag(s), {DepCount} distinct dependency ref(s)",
_driverInstanceId, _tagsByRawPath.Count, _dependencyRefs.Count);
return Task.CompletedTask;
}
///
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// A calc-tag add/remove/edit — or a script edit that changes the injected scriptSource — changes the
// merged DriverConfig, which DriverSpawnPlanner routes to an in-place ApplyDelta → THIS call (a
// respawn fires only on a DriverType/ResilienceConfig change). So the authored tag table is NOT
// fixed at construction: we re-bind the new config and REBUILD the tag table + dependency-ref set
// here. Without this, added tags never compute, removed tags keep publishing, and edited scripts
// serve stale results (deploy reports success but nothing changed). After the rebuild _dependencyRefs
// reflects the new authored tags, and the host re-registers the mux interest post-reinit (the
// DriverInstanceActor.DeltaApplied → DriverHostActor path, sequenced AFTER this completes).
StopTimers();
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
// recompile from each tag's NEW ScriptSource) + re-arm timers off the rebuilt tag set.
_evaluator.ClearCompiledScripts();
_running = true;
StartTimers();
return Task.CompletedTask;
}
///
public Task ShutdownAsync(CancellationToken cancellationToken)
{
_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;
}
///
/// Current health, derived from the driver's own lifecycle state.
/// Gitea #517. This used to return a hard-coded that
/// consulted nothing. That made it the fleet's one driver whose GetHealth() reported Healthy
/// with its timers stopped — not as a race window, but permanently and deterministically:
/// before , after , and after
/// . Every other driver reads a _health field that shutdown lands on
/// .
/// There is no transport to be Degraded about — this driver computes from the dependency mux,
/// so its only meaningful states are "running" and "not running". _running is the flag the
/// compute paths themselves gate on, which is what makes it the honest input here.
///
/// Healthy while running; otherwise Unknown, retaining the last compute time.
public DriverHealth GetHealth() => _running
? new DriverHealth(DriverState.Healthy, _lastComputeUtc, null)
: new DriverHealth(DriverState.Unknown, _lastComputeUtc, null);
// ---- IDependencyConsumer ----
///
public IReadOnlyCollection DependencyRefs => _dependencyRefs;
///
public void OnDependencyValue(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
{
if (!_running) return;
lock (_lock)
{
_depValues[rawPath] = new DataValueSnapshot(value, statusCode, timestampUtc, timestampUtc);
_arrivedDeps.Add(rawPath);
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)
{
if (PrepareEvaluation(calcRawPath, timestampUtc) is not { } prepared) continue;
_pending.Enqueue(prepared);
queued = true;
}
if (queued) SchedulePump();
}
}
// ---- ISubscribable ----
///
public Task SubscribeAsync(
IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var handle = new CalculationSubscriptionHandle($"calc-{Guid.NewGuid():N}");
// Initial-data callback (OPC UA convention): surface the current snapshot for each subscribed calc
// tag so a freshly-materialised node shows BadWaitingForInitialData until its first evaluation lands.
foreach (var reference in fullReferences)
{
var snapshot = ReadOne(reference);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, snapshot));
}
return Task.FromResult(handle);
}
///
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
=> Task.CompletedTask;
// ---- IReadable ----
///
public Task> ReadAsync(
IReadOnlyList fullReferences, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var results = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
results[i] = ReadOne(fullReferences[i]);
return Task.FromResult>(results);
}
private DataValueSnapshot ReadOne(string reference)
{
var now = DateTime.UtcNow;
lock (_lock)
{
// Gated on HAS PUBLISHED, not on "has a value": a tag whose only publish was an inputs-Bad
// report has no value of its own, and answering BadWaitingForInitialData for it would tell a
// polled reader "never observed" while the subscriber on the same node had been told
// BadNoCommunication. Something HAS been observed about such a tag — that its input is dead.
if (!_stateByRawPath.TryGetValue(reference, out var st) || !st.HasPublished)
return new DataValueSnapshot(null, StatusBadWaitingForInitialData, null, now);
// Exactly what the subscriber was last told — including a propagated input sub-code. Deriving
// it from a two-state flag here would make a read disagree with the subscription over the
// same node (05/STAB-19).
var status = st.LastPublishedStatus;
return new DataValueSnapshot(st.LastValue, status, st.LastTimestampUtc, st.LastTimestampUtc);
}
}
// ---- evaluation ----
/// 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' snapshots — value, status and timestamps — as of the trigger.
/// The trigger's timestamp, stamped on whatever is published.
/// The worst status among
/// (). Bad ⇒ the script is HELD and this status is published;
/// Good or Uncertain ⇒ the script runs and its result is stamped with it.
private sealed record PendingEvaluation(
CalculationTagDefinition Definition,
IReadOnlyDictionary Dependencies,
DateTime TimestampUtc,
uint WorstInputStatus);
///
/// 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 (!_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
var deps = new Dictionary(StringComparer.Ordinal);
foreach (var dep in def.DependencyRefs)
if (_depValues.TryGetValue(dep, out var v)) deps[dep] = v;
// Worst-of-input, computed from the SAME snapshot the evaluation will use, so the status
// published can never disagree with the values it was derived from (05/STAB-19).
var worst = InputQuality.WorstOf(def.DependencyRefs.Select(r => deps.GetValueOrDefault(r)));
return new PendingEvaluation(def, deps, timestampUtc, worst);
}
/// 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);
}
/// 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)
{
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;
// HOLD ON BAD (05/STAB-19, matching ScriptedAlarmEngine's AreInputsReady): a Bad input is not
// usable, so the script does NOT run — which removes the "(double)ctx.GetTag(x).Value threw"
// storm at its source rather than catching it downstream — and the tag reports the input's own
// sub-code. Uncertain IS usable: the script runs and the output inherits the Uncertain code.
if (!InputQuality.IsUsable(pending.WorstInputStatus))
{
(object? Value, uint StatusCode)? badPublish;
lock (_lock)
{
if (!_stateByRawPath.TryGetValue(def.RawPath, out var badState)) return;
badPublish = MarkInputsBad(badState, pending.WorstInputStatus, pending.TimestampUtc);
}
// No script log: an input dying is not a script fault, and the script log is the per-failure
// SCRIPT diagnostic. Here the status code is the diagnostic.
if (badPublish is { } bad) Publish(def.RawPath, bad.Value, bad.StatusCode, pending.TimestampUtc);
return;
}
// Snapshots, not bare values: the script's ctx.GetTag(path).StatusCode / .SourceTimestampUtc are
// then the input's real ones rather than a fabricated Good stamped at evaluation time (05/STAB-19).
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.WorstInputStatus, pending.TimestampUtc);
}
}
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 result is deduped.
/// The dedup key is (value, status), not value alone — 05/STAB-19 generalising the
/// Bad-recovery bypass 02/S13 introduced. Recovery is a publish, never an assumption: the node is
/// currently carrying the previous status, so a result whose VALUE is unchanged but whose STATUS
/// differs — the shape of every recovery, since a device that comes back usually comes back holding
/// the same reading — must still land, or the node stays Bad forever.
/// The tag's published state.
/// The computed value.
/// The status to stamp: the worst input status (Good, or an Uncertain sub-code).
/// The trigger's timestamp.
/// The snapshot to publish, or null when deduped.
private (object? Value, uint StatusCode)? MarkEvaluationSucceeded(
TagState st, object? value, uint statusCode, DateTime timestampUtc)
{
if (st.HasPublished && st.LastPublishedStatus == statusCode && st.HasValue && Equals(st.LastValue, value))
return null;
st.HasValue = true;
st.LastValue = value;
st.HasPublished = true;
st.LastPublishedStatus = statusCode;
st.LastTimestampUtc = timestampUtc;
_lastComputeUtc = DateTime.UtcNow;
return (value, statusCode);
}
/// Record that this tag's INPUTS are Bad, so no evaluation ran (05/STAB-19). Publishes the
/// worst input's status — carrying the last known value — once per change of that status: a
/// repeat of the same Bad is silent, while BadNoCommunication arriving over an earlier
/// BadInternalError (or the reverse) is not masked, because the sub-code is what names which
/// failure the operator is looking at. MUST be called under .
/// The tag's published state.
/// The worst status among the tag's declared inputs.
/// The trigger's timestamp.
/// The snapshot to publish, or null when that status was already published.
private static (object? Value, uint StatusCode)? MarkInputsBad(
TagState st, uint worstInputStatus, DateTime timestampUtc)
{
if (st.HasPublished && st.LastPublishedStatus == worstInputStatus) return null;
st.HasPublished = true;
st.LastPublishedStatus = worstInputStatus;
st.LastTimestampUtc = timestampUtc;
// Carry the last-known value (null if none) with the input's Bad status.
return (st.HasValue ? st.LastValue : null, worstInputStatus);
}
/// 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)
{
if (st.HasPublished && st.LastPublishedStatus == StatusBadInternalError) return null;
st.HasPublished = true;
st.LastPublishedStatus = StatusBadInternalError;
st.LastTimestampUtc = timestampUtc;
// Carry the last-known value (null if none) with Bad quality.
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 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));
}
private void PublishScriptLog(CalculationTagDefinition def, string reason)
{
// Route through the script logger bound with ScriptId + the calc tag's RawPath (in the VirtualTagId
// slot the Script-log page attributes on). The ScriptRootLogger's ScriptLogTopicSink converts this
// Warning event into a ScriptLogEntry on the script-logs DPS topic — the driver needs no Akka access.
_scriptRoot.Logger
.ForContext(ScriptLoggerFactory.ScriptIdProperty, def.ScriptId)
.ForContext(ScriptLoggerFactory.VirtualTagIdProperty, def.RawPath)
.Warning("{Reason}", reason);
}
private void StartTimers()
{
var timed = _tagsByRawPath.Values.Where(t => t.TimerInterval is not null).ToArray();
if (timed.Length == 0) return;
_timers = new CalculationTimerScheduler(EvaluateTagFromTimer, _logger);
_timers.Start(timed);
}
private void StopTimers()
{
_timers?.Dispose();
_timers = null;
}
/// 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.
private static readonly CalculationSubscriptionHandle SharedHandle = new("calc-shared");
///
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_running = false;
StopTimers();
lock (_lock) _pending.Clear();
_evaluator.Dispose();
}
/// Per-tag published state (guarded by ).
private sealed class TagState
{
public bool HasValue;
public object? LastValue;
/// Whether anything has been published for this tag yet. Distinct from
/// : an inputs-Bad publish carries no value of its own.
public bool HasPublished;
/// The status code last published for this tag. Replaces the old two-state
/// LastPublishedBad flag (05/STAB-19): the dedup key is (value, status), and a Bad status
/// now carries the failing INPUT's sub-code rather than a fixed BadInternalError.
public uint LastPublishedStatus;
public DateTime LastTimestampUtc;
}
private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
}