using System.Collections.Concurrent;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
///
/// Phase 7 scripted-alarm orchestrator. Compiles every configured alarm's predicate
/// against the Stream A sandbox, subscribes to the referenced upstream tags,
/// re-evaluates the predicate on every input change + on a shelving-check timer,
/// applies the resulting transition through ,
/// persists state via , and emits the resulting events
/// through (which wires into the existing
/// IAlarmSource fan-out).
///
///
///
/// Scripted alarms are leaves in the evaluation DAG — no alarm's state drives
/// another alarm's predicate. The engine maintains only an inverse index from
/// upstream tag path → alarms referencing it; no topological sort needed
/// (unlike the virtual-tag engine).
///
///
/// Evaluation errors (script throws, timeout, coercion fail) surface as
/// structured errors in the dedicated scripts-*.log sink plus a WARN companion
/// in the main log. The alarm's ActiveState stays at its prior value — the
/// engine does NOT invent a clear transition just because the predicate broke.
/// Operators investigating a broken predicate shouldn't see a phantom
/// clear-event preceding the failure.
///
///
/// Upstream pushes coalesce through a dirty set. A push does not spawn its own
/// re-evaluation: it marks the referencing alarm ids dirty and, if no drainer is already
/// running, starts one. then swaps the set out and evaluates
/// the batch, looping until the set is empty. N pushes arriving during one slow evaluation
/// therefore collapse into ONE further pass instead of N serialized ones that would each
/// re-read the same latest values. The collapse is loss-free by construction:
/// RefillReadCache always reads , which every push has
/// already updated, so the surviving pass evaluates the latest values — latest-wins, never
/// a lost update. There is no timer and no wall-clock batching window; a push that lands on
/// an idle engine is evaluated immediately, and the coalescing window is exactly "whatever
/// queued while was busy".
///
///
/// deliberately stays INSIDE the gate, before
/// the in-memory update. The persist-before-update ordering is a real crash-consistency
/// property — a store failure must leave the in-memory condition and the persisted one both
/// at the prior state rather than diverging — and it is pinned by
/// Store_save_failure_leaves_in_memory_state_unchanged. Post-coalescing a save only
/// happens on a genuine Part 9 transition, which is rare relative to the push rate, so the
/// gate-held store call is no longer on the hot path and a write-behind queue would trade a
/// correctness property for throughput the engine does not need.
///
///
public sealed class ScriptedAlarmEngine : IDisposable
{
///
/// OPC UA Part 4 BadWaitingForInitialData — the status the upstream source stamps on a
/// path nothing has ever pushed to, and the marker uses to tell
/// genuine cold start apart from a driver reporting Bad.
/// Inlined rather than taken from Commons.OpcUa.OpcUaStatus because this project
/// deliberately does not reference Commons — the engine is pure evaluation logic over
/// Core.Abstractions. The value is fixed by the OPC UA spec, so the duplication cannot
/// drift; DependencyMuxTagUpstreamSourceTests pins the two agreeing.
///
private const uint StatusNeverPushed = 0x80320000u;
private readonly ITagUpstreamSource _upstream;
private readonly IAlarmStateStore _store;
private readonly ScriptLoggerFactory _loggerFactory;
private readonly ILogger _engineLogger;
private readonly Func _clock;
private readonly TimeSpan _scriptTimeout;
///
/// Per-alarm run quarantine (keyed by alarm id, source identity = predicate source). Refuses a
/// predicate run while the alarm's previous (timed-out, abandoned) run is still executing and
/// quarantines after three consecutive timeouts (a refusal is not a strike unless its orphan is stuck past StuckOrphanQuarantineFactor × the budget) or an allocation-budget breach — one orphan
/// per alarm instead of one per upstream push. Reset on every (a redeploy).
///
private readonly ScriptRunGuard _guard;
///
/// lmxopcua#780 — the out-of-process runner when Scripting:Runner=OutOfProcess lists the ScriptedAlarm seam;
/// null (the default) keeps the in-process guard path, unchanged. With a runner the predicates compile and run in
/// the pre-emptible worker pool, and a re-evaluation pass submits its whole dirty set as one batch.
///
private readonly IScriptRunner? _runner;
private readonly IScriptShadow? _shadow;
// Alarms whose predicate came back Unavailable (the runner, not the script, failed) and that are waiting for the
// runner's Recovery to re-evaluate them once. Guarded by _dirtyLock.
private readonly HashSet _recoveryArmed = new(StringComparer.Ordinal);
// ConcurrentDictionary, not a plain Dictionary: every mutation happens under
// _evalGate, but four read paths (GetState, GetAllStates, LoadedAlarmIds,
// RunShelvingCheck) touch _alarms from arbitrary threads (Admin UI request
// threads, the shelving Timer thread-pool callback) without holding the gate.
// A plain Dictionary read concurrent with a writer's entry reassignment can
// throw or return torn state; ConcurrentDictionary makes entry assignment and
// snapshot enumeration safe. The only write shapes are indexer-set and Clear,
// both of which ConcurrentDictionary supports atomically.
private readonly ConcurrentDictionary _alarms = new(StringComparer.Ordinal);
///
/// Per-alarm reusable evaluation scratch. The read-cache dictionary and the
/// instance are both allocated once per
/// alarm (on first evaluation) and reused across every subsequent re-eval —
/// the hot path no longer allocates a fresh dictionary + context per upstream
/// tag change. Safe because only
/// runs under , which serialises every evaluation:
/// two threads can never observe the same scratch in a half-refilled state.
/// Cleared in alongside .
///
private readonly ConcurrentDictionary _scratchByAlarmId =
new(StringComparer.Ordinal);
///
/// Last emitted worst-of-input quality bucket per alarm (0 = Good, 1 = Uncertain,
/// 2 = Bad), computed each evaluation over the refilled read cache. A change in this bucket
/// with no accompanying Part 9 state transition drives a standalone
/// emission (a Bad input freezes the condition — no
/// transition — so quality can't ride one, exactly like a comms-lost native driver). Only ever
/// mutated under _evalGate; cleared alongside on load/dispose.
///
private readonly ConcurrentDictionary _lastQualityBucketByAlarmId =
new(StringComparer.Ordinal);
///
/// Compile cache for every alarm predicate. Routes 's
/// calls through the
/// cache so the collectible
/// each compile produces is actually disposed on the publish-replace path:
/// the cache's
/// disposes every materialised evaluator before dropping its dictionary entry,
/// so a config-publish releases the prior generation's ALCs and the per-publish
/// accretion the prior fix targeted is actually freed in production.
/// Pre-fix the engine called ScriptEvaluator.Compile directly, which left
/// the ALCs rooted until the process exited — defeating -008 on the real path.
///
private readonly CompiledScriptCache _compileCache = new();
///
/// Test-only diagnostic: returns the per-alarm scratch read-cache dictionary
/// if one has been allocated, else null. Used by regression tests to assert
/// the scratch is reused across evaluations (two reads return the same
/// instance).
///
///
/// Synchronization: the returned
/// is the engine's live mutable read-cache. It is refilled in place by
/// RefillReadCache on every predicate evaluation, under _evalGate.
/// Test callers MUST NOT iterate this dictionary while the engine is
/// actively evaluating (i.e. while an upstream change is mid-flight); the
/// refill clears the dict before repopulating and a concurrent iterator
/// would observe torn / partial state. Safe uses are: reference-identity
/// comparisons (e.g. asserting the same instance is reused across calls),
/// and single-key reads against an engine that has quiesced after a
/// deterministic upstream push. Anything more involved should snapshot a
/// copy under the gate.
///
/// The alarm identifier to look up.
/// The alarm's read-cache dictionary if one has been allocated; otherwise null.
internal IReadOnlyDictionary? TryGetScratchReadCacheForTest(string alarmId)
=> _scratchByAlarmId.TryGetValue(alarmId, out var s) ? s.ReadCache : null;
///
/// Test-only diagnostic: returns the per-alarm
/// if one has been allocated, else null. Companion to
/// .
///
///
/// Synchronization: the returned context wraps the same live
/// read-cache as — the same
/// "don't iterate during an in-flight evaluation" caveat applies. Safe
/// for reference-identity assertions on a quiesced engine.
///
/// The alarm identifier to look up.
/// The alarm's predicate context if one has been allocated; otherwise null.
internal AlarmPredicateContext? TryGetScratchContextForTest(string alarmId)
=> _scratchByAlarmId.TryGetValue(alarmId, out var s) ? s.Context : null;
private readonly ConcurrentDictionary _valueCache
= new(StringComparer.Ordinal);
/// Test hook — the number of paths currently held in the value cache.
internal int ValueCacheCountForTest => _valueCache.Count;
/// Test hook — whether is currently held in the value cache.
internal bool ValueCacheContainsForTest(string path) => _valueCache.ContainsKey(path);
///
/// How long waits for in-flight background work (a re-evaluation or shelving
/// check mid-SaveAsync) before cancelling it through and
/// returning. 5 s, like PollGroupEngine's drain; a hung store must not hang host shutdown.
///
internal TimeSpan DisposeDrainTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// Cancelled by once the drain bound expires — every background task
/// (DrainDirtyAsync, ShelvingCheckAsync) runs on its token, so a store call that honours
/// cancellation returns instead of outliving the engine.
private readonly CancellationTokenSource _shutdownCts = new();
/// 's token, captured once: background work reads this, never the CTS,
/// so can dispose the CTS without racing a late reader.
private readonly CancellationToken _shutdownToken;
private volatile bool _shutdownCtsDisposed;
/// Test hook — whether disposed .
internal bool ShutdownCtsDisposedForTest => _shutdownCtsDisposed;
private readonly Dictionary> _alarmsReferencing
= new(StringComparer.Ordinal); // tag path -> alarm ids
private readonly List _upstreamSubscriptions = [];
private readonly SemaphoreSlim _evalGate = new(1, 1);
private Timer? _shelvingTimer;
private bool _loaded;
// volatile: written by Dispose and read unsynchronized from every background
// path (LoadAsync's guard, OnUpstreamChange, the drain loop, ReevaluateAsync /
// ShelvingCheckAsync's post-gate re-checks, TrackBackgroundTask's registration
// check). Without it those reads may be hoisted or cached by the JIT, so a
// pushing thread can keep queueing work against a disposed engine.
private volatile bool _disposed;
// Tracks fire-and-forget background work launched by OnUpstreamChange
// (DrainDirtyAsync) and RunShelvingCheck (ShelvingCheckAsync). Dispose drains
// these so a re-evaluation in flight when shutdown begins finishes its
// SaveAsync before the engine returns control to the caller. The HashSet is
// accessed under its own lock — never under _evalGate — so registration /
// unregistration cannot deadlock against the gate.
private readonly HashSet _inFlight = [];
private readonly object _inFlightLock = new();
// Coalescing state for the upstream-push path. _dirtyAlarmIds accumulates the
// alarms a push made stale; _drainScheduled says a DrainDirtyAsync loop is
// already running and will pick them up, so a push storm starts exactly one
// drainer. Both are guarded by _dirtyLock — a leaf lock, never taken while
// holding _evalGate or _inFlightLock, and never held across an await.
private readonly HashSet _dirtyAlarmIds = new(StringComparer.Ordinal);
private readonly object _dirtyLock = new();
private bool _drainScheduled; // guarded by _dirtyLock
///
/// Initializes a new ScriptedAlarmEngine with the provided dependencies.
///
/// The upstream tag source for reading tag values.
/// The alarm state store for persistence.
/// The factory for creating alarm loggers.
/// The logger for engine-level diagnostics.
/// Optional function providing the current UTC time; defaults to DateTime.UtcNow.
/// Optional per-predicate wall-clock budget; null ⇒
/// of (2 s by default — the same budget virtual tags get). lmxopcua#779: the null
/// default used to be TimedScriptEvaluator.DefaultTimeout (250 ms), which is what every production engine
/// silently ran with because the host never passed a value.
/// Run-guard budget + quarantine threshold; null ⇒ defaults.
/// lmxopcua#780 — the out-of-process runner when the ScriptedAlarm seam runs out of
/// process; null (the default) keeps the in-process guard path, unchanged. Under Scripting:Runner=Shadow it is an
/// : the in-process path stays the one that drives the state machine and each predicate run
/// is mirrored to it for comparison (P2).
public ScriptedAlarmEngine(
ITagUpstreamSource upstream,
IAlarmStateStore store,
ScriptLoggerFactory loggerFactory,
ILogger engineLogger,
Func? clock = null,
TimeSpan? scriptTimeout = null,
ScriptingOptions? scripting = null,
IScriptRunner? scriptRunner = null)
{
_shadow = scriptRunner as IScriptShadow;
_runner = _shadow is null ? scriptRunner : null;
_upstream = upstream ?? throw new ArgumentNullException(nameof(upstream));
_store = store ?? throw new ArgumentNullException(nameof(store));
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
_engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger));
_clock = clock ?? (() => DateTime.UtcNow);
_scriptTimeout = scriptTimeout ?? (scripting ?? new ScriptingOptions()).Sanitized(out _).RunTimeout;
_guard = new ScriptRunGuard(scripting, _engineLogger);
_shutdownToken = _shutdownCts.Token;
}
/// The run guard every predicate evaluation goes through (diagnostics / tests).
public ScriptRunGuard RunGuard => _guard;
/// The wall-clock budget each predicate run is given (diagnostics / tests).
public TimeSpan ScriptTimeout => _scriptTimeout;
/// The out-of-process runner predicates run through, or null in process (diagnostics / tests).
public IScriptRunner? ScriptRunner => _runner;
/// The shadow predicate runs are mirrored to (Scripting:Runner=Shadow), or null.
public IScriptShadow? ScriptShadow => _shadow;
///
/// The alarm ids whose predicate is quarantined — by the in-process guard, or by the out-of-process runner when
/// one is in use (diagnostics / tests).
///
public IReadOnlyCollection QuarantinedAlarmIds => _runner is null
? _guard.QuarantinedKeys
: [.. _guard.QuarantinedKeys, .. _runner.QuarantinedKeys(ScriptContextKind.AlarmPredicate)];
///
/// OPC UA BadResourceUnavailable — the annotation for a predicate the out-of-process RUNNER could not run
/// (its worker restarted under it, or no worker was ready): the condition is frozen exactly as for a script
/// fault, but the quality tells the operator the node is sick, not the script (design §9, §14 #3; the same split
/// the virtual-tag and Calculation seams make).
///
private const uint StatusBadResourceUnavailable = 0x80040000u;
///
/// OPC UA BadInternalError — the quality a condition is annotated with while its predicate
/// cannot be evaluated (timed out, refused because the previous run is still executing, or
/// quarantined). The condition's state is frozen; the quality says why it cannot be trusted. The
/// next completed evaluation restores the input-derived quality through the normal bucket tracking.
///
private const uint StatusBadInternalError = 0x80020000u;
/// Raised for every emission the Part9StateMachine produces that the engine should publish.
public event EventHandler? OnEvent;
/// Gets the collection of loaded alarm identifiers.
public IReadOnlyCollection LoadedAlarmIds => _alarms.Keys.ToArray();
///
/// Load a batch of alarm definitions. Compiles every predicate, aggregates any
/// compile failures into one , subscribes
/// to upstream input tags, seeds the value cache, loads persisted state from
/// the store (falling back to Fresh for first-load alarms), and recomputes
/// ActiveState from the current predicate (startup recovery).
///
/// The alarm definitions to load.
/// The cancellation token.
/// A task that completes once every definition has been loaded and state restored.
public async Task LoadAsync(IReadOnlyList definitions, CancellationToken ct)
{
if (_disposed) throw new ObjectDisposedException(nameof(ScriptedAlarmEngine));
if (definitions is null) throw new ArgumentNullException(nameof(definitions));
var pending = new List(0);
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Stop the prior shelving timer BEFORE clearing _alarms so no in-progress
// tick can observe a half-cleared dictionary. Moving the dispose here also
// means a failed reload (compile errors, store failures) stops the timer
// even though execution never reaches the _shelvingTimer = new Timer(...)
// assignment at the bottom of the try block. Without this, the old timer
// keeps firing against the partially-cleared _alarms until Dispose() is
// eventually called — not a permanent leak, but an unexpected side effect
// during the window.
_shelvingTimer?.Dispose();
_shelvingTimer = null;
UnsubscribeFromUpstream();
_alarms.Clear();
_alarmsReferencing.Clear();
// The value cache is re-seeded below for exactly the paths this generation references. It
// used to survive reloads untouched, so every path a retired alarm had ever referenced stayed
// cached forever — a slow leak over many deploys with a churning tag set.
_valueCache.Clear();
// Drop the prior generation's per-alarm scratch buffers — definitions may
// have changed (different Inputs, different Logger), so any reuse would be
// unsafe.
_scratchByAlarmId.Clear();
_lastQualityBucketByAlarmId.Clear();
// Dispose every compiled-predicate ALC from the prior generation BEFORE we
// recompile this one. Skipping this is what made the earlier fix a
// no-op in production.
_compileCache.Clear();
// A (re)load is a deploy: lift every run-guard quarantine (an edited predicate already lifts
// its own through the guard's source identity; this covers "redeployed to retry").
_guard.ResetAll();
_runner?.Reset(ScriptContextKind.AlarmPredicate);
_shadow?.Reset(ScriptContextKind.AlarmPredicate);
lock (_dirtyLock) _recoveryArmed.Clear();
var compileFailures = new List();
foreach (var def in definitions)
{
try
{
var extraction = DependencyExtractor.Extract(def.PredicateScriptSource);
if (!extraction.IsValid)
{
var joined = string.Join("; ", extraction.Rejections.Select(r => r.Message));
compileFailures.Add($"{def.AlarmId}: dependency extraction rejected — {joined}");
continue;
}
// Route through CompiledScriptCache so the emitted assembly's
// collectible ALC participates in publish-replace cleanup.
// The compiled script is run through _guard (timeout + quarantine) at evaluation time —
// the per-alarm TimedScriptEvaluator this used to wrap gave the timeout but could not
// see that the run it abandoned was still executing.
//
// Out of process (lmxopcua#780) the worker compiles and runs the predicate; here the source is only
// compiled to PE bytes and dropped — same Roslyn, sandbox and diagnostics, so a broken predicate still
// fails the load exactly as in process — and no assembly is loaded into the host.
ScriptEvaluator? evaluator = null;
if (_runner is null) evaluator = _compileCache.GetOrCompile(def.PredicateScriptSource);
else _ = ScriptEvaluator.CompileToImage(def.PredicateScriptSource);
var logger = _loggerFactory.Create(def.AlarmId);
var templateTokens = MessageTemplate.ExtractTokenPaths(def.MessageTemplate);
var allInputs = new HashSet(extraction.Reads, StringComparer.Ordinal);
foreach (var t in templateTokens) allInputs.Add(t);
_alarms[def.AlarmId] = new AlarmState(def, evaluator, extraction.Reads, templateTokens, logger,
AlarmConditionState.Fresh(def.AlarmId, _clock()));
foreach (var path in allInputs)
{
if (!_alarmsReferencing.TryGetValue(path, out var set))
_alarmsReferencing[path] = set = new HashSet(StringComparer.Ordinal);
set.Add(def.AlarmId);
}
}
catch (Exception ex)
{
compileFailures.Add($"{def.AlarmId}: {ex.Message}");
}
}
if (compileFailures.Count > 0)
{
throw new InvalidOperationException(
$"ScriptedAlarmEngine load failed. {compileFailures.Count} alarm(s) did not compile:\n "
+ string.Join("\n ", compileFailures));
}
// Seed the value cache with current tag values before subscribing. The
// ReadTag calls happen first so that the initial predicate evaluation below
// (startup recovery) uses a consistent snapshot.
// Subscriptions are established AFTER _loaded = true so that any synchronous
// initial-push an ITagUpstreamSource delivers from inside SubscribeTag arrives
// when _alarms is fully initialised. Before _loaded = true, a synchronous push
// would race the in-progress state restore and could overwrite the carefully
// seeded cache with a push that has no defined ordering relative to ReadTag.
foreach (var path in _alarmsReferencing.Keys)
_valueCache[path] = _upstream.ReadTag(path);
// Restore persisted state, falling back to Fresh where nothing was saved,
// then re-derive ActiveState from the current predicate.
// Any predicate emissions queue into `pending` and fire after the gate
// is released — so a startup-recovery activation event can call back into
// the engine without deadlocking.
if (_runner is null)
{
foreach (var (alarmId, state) in _alarms)
{
var persisted = await _store.LoadAsync(alarmId, ct).ConfigureAwait(false);
var seed = persisted ?? state.Condition;
var afterPredicate = await EvaluatePredicateToStateAsync(state, seed, nowUtc: _clock(), ct, pending)
.ConfigureAwait(false);
_alarms[alarmId] = state with { Condition = afterPredicate };
await _store.SaveAsync(afterPredicate, ct).ConfigureAwait(false);
}
}
else
{
// Out of process: every predicate goes in one batch, then each alarm is applied and saved in the
// same order the serial loop above uses.
var seeds = new List<(AlarmState State, AlarmConditionState Seed)>(_alarms.Count);
foreach (var (alarmId, state) in _alarms)
seeds.Add((state, await _store.LoadAsync(alarmId, ct).ConfigureAwait(false) ?? state.Condition));
var after = await EvaluatePredicateBatchToStatesAsync(seeds, _clock(), ct, pending).ConfigureAwait(false);
for (var i = 0; i < seeds.Count; i++)
{
var state = seeds[i].State;
_alarms[state.Definition.AlarmId] = state with { Condition = after[i] };
await _store.SaveAsync(after[i], ct).ConfigureAwait(false);
}
}
_loaded = true;
// Subscribe after _loaded = true and full state restore. If an upstream
// implementation pushes its initial value synchronously from inside
// SubscribeTag, OnUpstreamChange will queue a ReevaluateAsync that acquires
// _evalGate — it will correctly block until LoadAsync releases the gate, then
// re-evaluate against the fully-populated _alarms dict.
foreach (var path in _alarmsReferencing.Keys)
_upstreamSubscriptions.Add(_upstream.SubscribeTag(path, OnUpstreamChange));
_engineLogger.Information("ScriptedAlarmEngine loaded {Count} alarm(s)", _alarms.Count);
// Start the shelving-check timer — ticks every 5s, expires any timed shelves
// that have passed their UnshelveAtUtc. The prior timer was already disposed
// at the START of this try block, so _shelvingTimer
// is null here; no double-dispose risk.
_shelvingTimer = new Timer(_ => RunShelvingCheck(),
null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5));
}
finally
{
_evalGate.Release();
}
// Fire any emissions collected during startup recovery OUTSIDE the gate so
// subscribers can re-enter the engine safely.
foreach (var evt in pending) FireEvent(evt);
}
///
/// Current persisted state for . Returns null for
/// unknown alarm. Mainly used for diagnostics + the Admin UI status page.
///
/// The alarm identifier.
/// The alarm's current condition state, or null if the alarm id is unknown.
public AlarmConditionState? GetState(string alarmId)
=> _alarms.TryGetValue(alarmId, out var s) ? s.Condition : null;
/// Gets the current persisted state for all loaded alarms.
/// The current condition state of every loaded alarm.
public IReadOnlyCollection GetAllStates()
=> _alarms.Values.Select(a => a.Condition).ToArray();
///
/// The currently-held condition of every loaded alarm, packaged with the definition's
/// severity, the rendered message template and the last-observed input quality, i.e. everything
/// a caller needs to project the alarm onto an OPC UA condition node without waiting for
/// the next transition.
///
///
///
/// This is a read, not an emission. It deliberately returns a distinct type rather
/// than a , and it never touches . A
/// still-active alarm produces at load (there is no
/// transition to report), so the transition stream cannot carry the current state — that is
/// exactly why the caller needs this. Routing these through the emission path instead would
/// append a duplicate historian / alerts row on every deploy, so the shape here is
/// intentionally one the emission machinery cannot consume.
///
///
/// Synchronization. Reads _alarms and _valueCache — both
/// — without taking _evalGate, the
/// same posture as / . Each projection is
/// individually coherent; the set as a whole is not an atomic snapshot across a concurrent
/// re-evaluation. That is sufficient for the node-projection use: an evaluation racing this
/// read publishes its own transition through the normal path.
///
///
/// One projection per loaded alarm; empty before the first .
public IReadOnlyList GetProjections()
{
var projections = new List(_alarms.Count);
foreach (var (alarmId, state) in _alarms)
{
projections.Add(new ScriptedAlarmProjection(
AlarmId: alarmId,
Severity: state.Definition.Severity,
Message: MessageTemplate.Resolve(state.Definition.MessageTemplate, TryLookup),
Condition: state.Condition,
WorstInputStatusCode: LastWorstStatus(alarmId)));
}
return projections;
}
/// Acknowledges the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the acknowledgment.
/// An optional comment to attach to the acknowledgment.
/// The cancellation token.
/// A task that completes once the acknowledgment has been applied and persisted.
public Task AcknowledgeAsync(string alarmId, string user, string? comment, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyAcknowledge(cur, user, comment, _clock()));
/// Confirms the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the confirmation.
/// An optional comment to attach to the confirmation.
/// The cancellation token.
/// A task that completes once the confirmation has been applied and persisted.
public Task ConfirmAsync(string alarmId, string user, string? comment, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyConfirm(cur, user, comment, _clock()));
/// Applies a one-shot shelve to the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the shelve operation.
/// The cancellation token.
/// A task that completes once the shelve has been applied and persisted.
public Task OneShotShelveAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyOneShotShelve(cur, user, _clock()));
/// Applies a timed shelve to the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the shelve operation.
/// The UTC time at which the shelve will automatically expire.
/// The cancellation token.
/// A task that completes once the timed shelve has been applied and persisted.
public Task TimedShelveAsync(string alarmId, string user, DateTime unshelveAtUtc, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyTimedShelve(cur, user, unshelveAtUtc, _clock()));
/// Removes any shelve from the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the unshelve operation.
/// The cancellation token.
/// A task that completes once the unshelve has been applied and persisted.
public Task UnshelveAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyUnshelve(cur, user, _clock()));
/// Enables the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the enable operation.
/// The cancellation token.
/// A task that completes once the enable has been applied and persisted.
public Task EnableAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyEnable(cur, user, _clock()));
/// Disables the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user performing the disable operation.
/// The cancellation token.
/// A task that completes once the disable has been applied and persisted.
public Task DisableAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyDisable(cur, user, _clock()));
/// Adds a comment to the specified alarm on behalf of the given user.
/// The alarm identifier.
/// The user adding the comment.
/// The comment text.
/// The cancellation token.
/// A task that completes once the comment has been recorded and persisted.
public Task AddCommentAsync(string alarmId, string user, string text, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyAddComment(cur, user, text, _clock()));
private async Task ApplyAsync(string alarmId, CancellationToken ct, Func op)
{
EnsureLoaded();
if (!_alarms.TryGetValue(alarmId, out var state))
throw new ArgumentException($"Unknown alarm {alarmId}", nameof(alarmId));
ScriptedAlarmEvent? pending = null;
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var result = op(state.Condition);
// Persist BEFORE updating in-memory so a store failure leaves both
// in-memory and persisted at the prior state rather than diverging.
// If SaveAsync throws the in-memory _alarms entry stays unchanged and
// the exception propagates to the caller.
await _store.SaveAsync(result.State, ct).ConfigureAwait(false);
_alarms[alarmId] = state with { Condition = result.State };
// Build the emission event under the gate (it captures a coherent
// snapshot of state + message-template values) but defer the actual
// OnEvent dispatch until after Release() so a slow subscriber or a
// subscriber that re-enters the engine doesn't block / deadlock.
if (result.Emission != EmissionKind.None)
pending = BuildEmission(state, result.State, result.Emission, LastWorstStatus(alarmId));
else if (result.NoOpReason is { } reason)
{
// The Part9StateMachine remarks promise a diagnostic log line for
// disabled-alarm no-ops + idempotent ack/confirm/shelve/unshelve
// calls. We surface them at debug so they're available when
// investigating "why didn't my ack take effect?" without spamming
// the main info log.
state.Logger.Debug("Alarm {AlarmId} no-op transition: {Reason}", alarmId, reason);
}
}
finally { _evalGate.Release(); }
// OnEvent dispatch happens OUTSIDE _evalGate so subscribers can call back
// into the engine (e.g. AcknowledgeAsync from inside an Activated handler)
// without deadlocking against the non-reentrant SemaphoreSlim.
if (pending is not null) FireEvent(pending);
}
///
/// Upstream-change callback. Updates the value cache + marks every alarm
/// referencing the changed path dirty, starting the drainer if one isn't
/// already running. Never blocks driver-side dispatch, and never spawns work
/// per push: a push landing while a drain is in flight only grows the dirty
/// set, so a storm collapses into one further evaluation pass over the latest
/// values. The drainer is tracked so can drain it.
///
/// The upstream tag path that changed.
/// The new data value snapshot.
internal void OnUpstreamChange(string path, DataValueSnapshot value)
{
if (_disposed) return; // a disposed engine neither caches nor queues — nothing reads it again
_valueCache[path] = value;
if (!_alarmsReferencing.TryGetValue(path, out var alarmIds)) return;
MarkDirty(alarmIds);
}
/// Marks dirty and starts the drainer if none is running.
private void MarkDirty(IEnumerable alarmIds)
{
bool startDrain;
lock (_dirtyLock)
{
// Iterate the referencing set directly — the pre-coalescing path
// allocated a fresh string[] on every single push.
foreach (var id in alarmIds) _dirtyAlarmIds.Add(id);
startDrain = !_drainScheduled;
if (startDrain) _drainScheduled = true;
}
// Started outside the lock: DrainDirtyAsync runs synchronously up to its
// first real await, and nothing it does should happen under _dirtyLock.
if (startDrain) TrackBackgroundTask(() => DrainDirtyAsync(_shutdownToken));
}
///
/// The single drainer for the upstream-push dirty set. Swaps the accumulated
/// alarm ids out under _dirtyLock and re-evaluates that batch, looping
/// until the set is empty — so every push that landed while the previous pass
/// held _evalGate is served by the next pass rather than by a queue of
/// redundant per-push evaluations. Exactly one instance runs at a time
/// (_drainScheduled), and it clears that flag under the same lock it
/// reads the set under, so the last drainer out and the next push in can never
/// both decide "someone else will handle it".
///
///
/// INVARIANT: _drainScheduled is released on EVERY exit path, the
/// exceptional one included. The flag is the only thing that lets
/// start a drainer, so an escape that leaves it
/// stuck true permanently and silently stops evaluation for every scripted
/// alarm in the process: pushes keep filling _valueCache and the dirty set,
/// no drainer is ever started again, no predicate ever runs again, and nothing is
/// logged. catches everything today, so the throw
/// surface is currently empty — the finally below is what keeps the
/// invariant true when it stops being (an audit-mode Serilog sink rethrowing out
/// of that catch block's own Error() call is enough).
///
private async Task DrainDirtyAsync(CancellationToken ct)
{
// Set inside _dirtyLock, together with the flag clear, so the finally can tell
// "I already released" from "I am unwinding" without ever double-releasing —
// a second clear could stomp a _drainScheduled = true a racing push had just
// set for the drainer it started, leaving two drainers running.
var released = false;
try
{
while (true)
{
string[] batch;
lock (_dirtyLock)
{
// Bail promptly on a disposing engine, but ALWAYS release
// _drainScheduled on the way out — leaving it stuck true would make a
// hypothetical post-dispose push believe a drainer is coming. (Such
// pushes are already dropped in OnUpstreamChange; this keeps the flag
// honest regardless.)
if (_disposed || _dirtyAlarmIds.Count == 0)
{
_dirtyAlarmIds.Clear();
_drainScheduled = false;
released = true;
return;
}
batch = [.. _dirtyAlarmIds];
_dirtyAlarmIds.Clear();
}
// Reused as-is: the gate / evaluate / persist / emit-outside-the-gate shape
// is already correct for a batch. Only the spawn-per-push is gone.
await ReevaluateAsync(batch, ct).ConfigureAwait(false);
}
}
finally
{
// Exceptional exit only (see `released`). The batch the faulted pass was
// carrying dies with it — ReevaluateAsync's own catch is what normally makes
// that a logged non-event — but ids that queued BEHIND it are still in the
// set and would otherwise sit there until some later push happened to arrive,
// so hand the latch straight to a replacement drainer instead of releasing it.
//
// Exactly-one-drainer survives the handover because _drainScheduled never
// transiently CLEARS during one: deciding to hand over and clearing the flag
// are the same guarded read-modify-write, and the clear happens only in the
// branch that starts no replacement. The spawn itself is deliberately OUTSIDE
// the lock (nothing it does belongs under _dirtyLock) and is safe there
// precisely because the flag is still set — a push landing in that window only
// grows the set, which the replacement then drains.
//
// Two bounds, both honest:
// * CPU — every pass swaps the set out and clears it before evaluating, so a
// replacement finds work only if fresh pushes arrived. A permanently
// throwing evaluation therefore costs one pass per push, the same rate as
// a healthy one; it cannot spin against an idle upstream.
// * Stack — the replacement is QUEUED (Task.Run), not called. A replacement
// that is called runs INLINE for as long as it does not truly suspend, and
// its own escape then unwinds back into this same finally block, nesting a
// frame per throw with no ceiling — a StackOverflowException no catch can
// recover from. Whether it suspends first is not a property this method
// controls: _evalGate is already free when the escape is thrown (released
// in ReevaluateAsync's inner finally, before the outer catch's logging
// call), so a failure raised BEFORE the first awaited work — the clock
// read, anything synchronous ahead of the predicate — unwinds with no
// suspension at all. A failure raised at or after the predicate does not,
// because TimedScriptEvaluator hops to the pool on every evaluation;
// measured, that path holds depth flat even when called directly. Queuing
// makes the fresh stack unconditional instead of dependent on which throw
// shape happens to occur, at the price of one pool dispatch per faulted
// pass. Pinned by
// Sustained_synchronous_drain_failures_hand_over_without_deepening_the_stack.
if (!released)
{
bool restart;
lock (_dirtyLock)
{
restart = !_disposed && _dirtyAlarmIds.Count > 0;
if (!restart) _drainScheduled = false;
}
// Task.Run stays INSIDE the factory: the queued-not-called handover is what
// keeps the replacement off this unwinding stack (see the Stack bound above).
if (restart) TrackBackgroundTask(() => Task.Run(() => DrainDirtyAsync(ct)));
}
}
}
private async Task ReevaluateAsync(IReadOnlyList alarmIds, CancellationToken ct)
{
var pending = new List(0);
try
{
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Re-check after acquiring the gate: a Dispose() call may have
// completed between our _evalGate.WaitAsync and here. Writing to a
// disposing store or mutating _alarms after clear is unsafe.
if (_disposed) return;
if (_runner is not null)
{
// Out of process (lmxopcua#780, design §4.5): the whole dirty set in one batch, applied in order.
await ReevaluateBatchAsync(alarmIds, ct, pending).ConfigureAwait(false);
}
else
{
foreach (var id in alarmIds)
{
if (!_alarms.TryGetValue(id, out var state)) continue;
var newState = await EvaluatePredicateToStateAsync(
state, state.Condition, _clock(), ct, pending).ConfigureAwait(false);
if (!ReferenceEquals(newState, state.Condition))
{
// Persist before updating in-memory so a store failure leaves
// both sides at the prior state.
await _store.SaveAsync(newState, ct).ConfigureAwait(false);
_alarms[id] = state with { Condition = newState };
}
}
}
}
finally { _evalGate.Release(); }
}
catch (OperationCanceledException) when (_shutdownToken.IsCancellationRequested)
{
// Dispose cut the drain short after its bound — expected, not a failure.
return;
}
catch (Exception ex)
{
_engineLogger.Error(ex, "ScriptedAlarmEngine reevaluate failed");
return;
}
// Fire emissions OUTSIDE _evalGate so subscriber callbacks can re-enter
// the engine without deadlocking.
foreach (var evt in pending) FireEvent(evt);
}
///
/// Evaluate the predicate + apply the resulting state-machine transition.
/// Returns the new condition state. If the transition produces an emission,
/// appends it to so the caller can fire
/// them after releasing _evalGate — keeping subscriber callbacks
/// outside the gate.
///
///
/// Every caller (LoadAsync and ReevaluateAsync) owns a pending list and
/// passes it here; emissions are always deferred to after the gate is released.
/// The parameter is required (non-nullable) to make this contract explicit and
/// prevent a future caller from accidentally firing events under the gate.
///
private async Task EvaluatePredicateToStateAsync(
AlarmState state, AlarmConditionState seed, DateTime nowUtc, CancellationToken ct,
List pendingEmissions)
{
// Look up (or lazily allocate) the per-alarm scratch and refill its read cache
// in place. The dictionary + context survive across evaluations so the hot path
// no longer allocates per upstream tag change.
var scratch = _scratchByAlarmId.GetOrAdd(
state.Definition.AlarmId,
_ => new AlarmScratch(state.Inputs, state.Logger, _clock));
RefillReadCache(scratch.ReadCache, state.Inputs);
// Worst OPC UA quality across the alarm's inputs, computed BEFORE the readiness
// short-circuit so an outright-Bad input is still observed. A bucket change with no state
// transition is delivered out of band as a QualityChanged emission (see below).
var worstStatus = WorstInputStatus(scratch.ReadCache);
// Cold-start guard — skip the predicate when any referenced upstream tag has no
// cached value yet (the upstream subscription hasn't delivered its first push).
// Without this, predicates that cast `(double)ctx.GetTag(path).Value` throw NRE on
// every tick until the cache fills, spamming the log with identical stack traces.
// Bad quality is treated the same: the input isn't available at the predicate's
// expected type, so the only defensible move is to hold the prior condition state.
if (!AreInputsReady(scratch.ReadCache))
{
// The condition is frozen (can't trust its state), but its source quality just changed
// bucket — annotate it out of band so a comms-lost / Bad-input scripted condition reports
// Bad, mirroring the native OT path.
if (TrackQualityBucket(state.Definition.AlarmId, worstStatus))
pendingEmissions.Add(BuildQualityEmission(state, seed, worstStatus));
return seed;
}
var context = scratch.Context;
bool predicateTrue;
try
{
predicateTrue = await _guard.RunAsync(
state.Definition.AlarmId, state.Definition.PredicateScriptSource, state.Evaluator!, context,
_scriptTimeout, ct)
.ConfigureAwait(false);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
}
catch (ScriptTimeoutException tex)
{
MirrorPredicate(state, scratch, ScriptRunOutcome.Failed(
$"{ZB.MOM.WW.OtOpcUa.Commons.Engines.ScriptEvalFailureReasons.TimedOutPrefix}{tex.Timeout.TotalSeconds:F1}s"));
return PredicateDidNotComplete(state, seed, scratch,
$"timed out after {tex.Timeout.TotalMilliseconds:F0} ms", pendingEmissions);
}
catch (ScriptRunRefusedException ex)
{
MirrorPredicate(state, scratch, ScriptRunOutcome.Failed(ex.Message));
return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions);
}
catch (ScriptQuarantinedException ex)
{
MirrorPredicate(state, scratch, ScriptRunOutcome.Failed(ex.Message));
return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions);
}
catch (Exception ex)
{
MirrorPredicate(state, scratch, ScriptRunOutcome.Failed($"script threw: {ex.Message}"));
scratch.LastNonCompletion = null;
state.Logger.Error(ex, "Alarm predicate threw — state unchanged");
return seed;
}
MirrorPredicate(state, scratch, ScriptRunOutcome.Ok(predicateTrue));
scratch.LastNonCompletion = null;
// Track the bucket only once the predicate has COMPLETED (I3). The tracked bucket is the last
// bucket this alarm was annotated with; tracking the input bucket before the attempt flipped it to
// Good on every push, so a refused / quarantined predicate (which annotates Bad) re-emitted
// QualityChanged on every push instead of once per real bucket change.
var qualityBucketChanged = TrackQualityBucket(state.Definition.AlarmId, worstStatus);
var result = Part9StateMachine.ApplyPredicate(seed, predicateTrue, nowUtc);
var transition = result.Emission != EmissionKind.None
? BuildEmission(state, result.State, result.Emission, worstStatus)
: null;
if (transition is not null)
{
// A real transition carries the current worst quality so the projected full-snapshot
// write doesn't clobber quality back to Good (e.g. a transition while an input is Uncertain).
pendingEmissions.Add(transition);
}
else if (qualityBucketChanged)
{
// No transition (or a Suppressed one) but the quality bucket moved — annotate out of band.
pendingEmissions.Add(BuildQualityEmission(state, result.State, worstStatus));
}
return result.State;
}
///
/// lmxopcua#780 P2 — Shadow: mirrors one in-process predicate run, after it has driven the state machine's input,
/// to the shadow runner for comparison. Never blocks; the inputs are copied before it returns, so the alarm's
/// reused read cache can be refilled at once.
///
private void MirrorPredicate(AlarmState state, AlarmScratch scratch, ScriptRunOutcome published)
{
if (_shadow is null) return;
_shadow.Mirror(new ScriptInvocation(
state.Definition.AlarmId, ScriptContextKind.AlarmPredicate, state.Definition.PredicateScriptSource,
scratch.ReadCache, _scriptTimeout), published);
}
///
/// The out-of-process re-evaluation pass (lmxopcua#780, design §4.5): every dirty alarm's predicate goes to the
/// runner in ONE batch, then the results are applied — Part 9 transition, persist-before-update, emissions — in
/// the original order, under _evalGate. Predicates are pure functions of the input snapshot taken here,
/// under the gate, so the state machine sees exactly the sequence the serial loop produces. The batch costs one
/// round trip for the whole set instead of one per alarm, which is what keeps a large engine inside its budget.
///
private async Task ReevaluateBatchAsync(IReadOnlyList alarmIds, CancellationToken ct, List pending)
{
var items = new List<(AlarmState State, AlarmConditionState Seed)>(alarmIds.Count);
foreach (var id in alarmIds)
if (_alarms.TryGetValue(id, out var state)) items.Add((state, state.Condition));
if (items.Count == 0) return;
var after = await EvaluatePredicateBatchToStatesAsync(items, _clock(), ct, pending).ConfigureAwait(false);
for (var i = 0; i < items.Count; i++)
{
var (state, seed) = items[i];
if (ReferenceEquals(after[i], seed)) continue;
// Persist before updating in-memory so a store failure leaves both sides at the prior state.
await _store.SaveAsync(after[i], ct).ConfigureAwait(false);
_alarms[state.Definition.AlarmId] = state with { Condition = after[i] };
}
}
///
/// for a batch, through : the same readiness gate,
/// quality tracking and Part 9 application per alarm, in order, with every predicate that has to run submitted
/// together. Failure semantics (design §9): a script fault, timeout, kill or quarantine freezes the condition with a
/// BadInternalError annotation, exactly as in process; a runner failure (worker restarted, none ready) freezes
/// it with BadResourceUnavailable and re-evaluates it once the runner can run it again. A dead worker never
/// clears an alarm: no outcome but a completed bool reaches the state machine.
///
private async Task EvaluatePredicateBatchToStatesAsync(
IReadOnlyList<(AlarmState State, AlarmConditionState Seed)> items, DateTime nowUtc, CancellationToken ct,
List pendingEmissions)
{
var runner = _runner!;
var after = new AlarmConditionState[items.Count];
var scratches = new AlarmScratch[items.Count];
var worst = new uint[items.Count];
var slot = new int[items.Count];
var invocations = new List(items.Count);
// Pass 1, no side effects: snapshot each alarm's inputs and decide whether its predicate runs.
for (var i = 0; i < items.Count; i++)
{
var state = items[i].State;
var scratch = scratches[i] = _scratchByAlarmId.GetOrAdd(
state.Definition.AlarmId, _ => new AlarmScratch(state.Inputs, state.Logger, _clock));
RefillReadCache(scratch.ReadCache, state.Inputs);
worst[i] = WorstInputStatus(scratch.ReadCache);
if (!AreInputsReady(scratch.ReadCache))
{
slot[i] = -1;
continue;
}
slot[i] = invocations.Count;
// A copy: the runner may still read the inputs after the caller has its answer (a soft timeout), and the
// scratch is refilled by the next pass.
invocations.Add(new ScriptInvocation(state.Definition.AlarmId, ScriptContextKind.AlarmPredicate,
state.Definition.PredicateScriptSource,
new Dictionary(scratch.ReadCache, StringComparer.Ordinal), _scriptTimeout, state.Logger));
}
var outcomes = invocations.Count == 0 ? [] : runner.RunBatch(invocations, ct);
PredicateBatchSubmittedForTest?.Invoke(invocations.Count);
// Pass 2, in order: exactly what the serial path does with each alarm.
for (var i = 0; i < items.Count; i++)
{
var (state, seed) = items[i];
var scratch = scratches[i];
if (slot[i] < 0)
{
if (TrackQualityBucket(state.Definition.AlarmId, worst[i]))
pendingEmissions.Add(BuildQualityEmission(state, seed, worst[i]));
after[i] = seed;
continue;
}
var outcome = await outcomes[slot[i]].ConfigureAwait(false);
after[i] = ApplyPredicateOutcome(state, seed, scratch, worst[i], outcome, nowUtc, ct, pendingEmissions);
}
return after;
}
/// One runner outcome onto one alarm (the second half of ).
private AlarmConditionState ApplyPredicateOutcome(AlarmState state, AlarmConditionState seed, AlarmScratch scratch,
uint worstStatus, ScriptRunOutcome outcome, DateTime nowUtc, CancellationToken ct, List pendingEmissions)
{
switch (outcome.Status)
{
case ScriptRunStatus.Unavailable:
// Not the script's fault: frozen, annotated BadResourceUnavailable, and evaluated again (against the
// inputs current then) when its worker is back.
ArmRecovery(state.Definition.AlarmId, outcome.Recovery);
return PredicateDidNotComplete(state, seed, scratch, outcome.Reason ?? ScriptRunOutcome.RunnerUnavailableReason,
pendingEmissions, StatusBadResourceUnavailable);
case ScriptRunStatus.Failed:
var reason = outcome.Reason ?? "evaluator failure";
if (ct.IsCancellationRequested) throw new OperationCanceledException(ct);
if (reason.StartsWith(ScriptThrewPrefix, StringComparison.Ordinal))
{
// As in process: a predicate that threw logs an Error and keeps its state, with no quality annotation.
scratch.LastNonCompletion = null;
state.Logger.Error("Alarm predicate threw — state unchanged: {Reason}", reason[ScriptThrewPrefix.Length..]);
return seed;
}
// Timed out, killed, refused, quarantined, unsupported type, did not compile: frozen + BadInternalError.
return PredicateDidNotComplete(state, seed, scratch, reason, pendingEmissions);
}
if (outcome.Value is not bool predicateTrue)
{
scratch.LastNonCompletion = null;
state.Logger.Error("Alarm predicate returned {Type} instead of a bool — state unchanged", outcome.Value?.GetType().Name ?? "null");
return seed;
}
scratch.LastNonCompletion = null;
// From here on, identical to the serial path: the bucket is tracked only once the predicate completed (I3).
var qualityBucketChanged = TrackQualityBucket(state.Definition.AlarmId, worstStatus);
var result = Part9StateMachine.ApplyPredicate(seed, predicateTrue, nowUtc);
var transition = result.Emission != EmissionKind.None
? BuildEmission(state, result.State, result.Emission, worstStatus)
: null;
if (transition is not null) pendingEmissions.Add(transition);
else if (qualityBucketChanged) pendingEmissions.Add(BuildQualityEmission(state, result.State, worstStatus));
return result.State;
}
/// The prefix the runners put on a predicate that threw (the in-process path catches the exception instead).
private const string ScriptThrewPrefix = "script threw: ";
///
/// Re-evaluate once when completes — the runner's worker is
/// back (design §9 "re-evaluated on return"). At most one wait per alarm.
///
private void ArmRecovery(string alarmId, Task? recovery)
{
if (recovery is null) return;
lock (_dirtyLock)
{
if (_disposed || !_recoveryArmed.Add(alarmId)) return;
}
_ = recovery.ContinueWith(_ =>
{
lock (_dirtyLock) _recoveryArmed.Remove(alarmId);
if (!_disposed) MarkDirty([alarmId]);
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
/// Test hook: the number of predicates each out-of-process batch pass submitted.
internal Action? PredicateBatchSubmittedForTest { get; set; }
///
/// The predicate did not return (timed out, refused because the previous run is still executing, or
/// quarantined): hold the condition at , log the reason once per change of
/// reason (a quarantined alarm is re-attempted on every push and would otherwise log every time),
/// and annotate the condition out of band — the frozen state
/// cannot be trusted and the quality is how the operator sees that. The next completed evaluation
/// restores the input-derived quality through the normal bucket tracking.
///
private AlarmConditionState PredicateDidNotComplete(
AlarmState state, AlarmConditionState seed, AlarmScratch scratch, string reason,
List pendingEmissions, uint status = StatusBadInternalError)
{
if (!string.Equals(scratch.LastNonCompletion, reason, StringComparison.Ordinal))
{
scratch.LastNonCompletion = reason;
state.Logger.Warning("Alarm predicate did not complete ({Reason}) — state unchanged, quality Bad", reason);
}
if (TrackQualityBucket(state.Definition.AlarmId, status))
pendingEmissions.Add(BuildQualityEmission(state, seed, status));
return seed;
}
///
/// Refill in place from _valueCache, falling
/// back to a synchronous ITagUpstreamSource.ReadTag for paths whose
/// first upstream push hasn't arrived yet. The dictionary is cleared and
/// repopulated under _evalGate so no concurrent reader can observe
/// a partial state. Replaces the old BuildReadCache which allocated a
/// fresh dictionary every call.
///
private void RefillReadCache(
Dictionary cache, IReadOnlySet inputs)
{
cache.Clear();
foreach (var p in inputs)
cache[p] = _valueCache.TryGetValue(p, out var v) ? v : _upstream.ReadTag(p);
}
///
/// Returns true when every entry in has a non-null value
/// and a Good-quality . A false here lets
/// callers short-circuit script evaluation — predicates that unconditionally cast
/// ctx.GetTag(path).Value to a numeric type would otherwise throw
/// until the upstream subscription delivers
/// its first push.
///
private static bool AreInputsReady(IReadOnlyDictionary cache)
{
foreach (var kv in cache)
{
if (kv.Value is null || kv.Value.Value is null) return false;
// OPC UA Part 4 StatusCode: bit 31 = severity 10 (Bad). Treat Good + Uncertain
// as "ready"; Uncertain carries a value the script can still inspect + make a
// qualified decision from. Only outright Bad is skipped.
if ((kv.Value.StatusCode & 0x80000000u) != 0) return false;
}
return true;
}
///
/// Build (but do not fire) the for a
/// transition. Returns null for kinds that should not be published
/// ( and
/// ). Pure construction — called under
/// _evalGate so the message-template resolution uses a coherent
/// value-cache snapshot. The actual dispatch is
/// done by AFTER the gate is
/// released.
///
private ScriptedAlarmEvent? BuildEmission(
AlarmState state, AlarmConditionState condition, EmissionKind kind, uint worstInputStatus)
{
// Suppressed kind means shelving ate the emission — we don't fire for subscribers
// but the state record still advanced so startup recovery reflects reality.
if (kind == EmissionKind.Suppressed || kind == EmissionKind.None) return null;
var message = MessageTemplate.Resolve(state.Definition.MessageTemplate, TryLookup);
return new ScriptedAlarmEvent(
AlarmId: state.Definition.AlarmId,
EquipmentPath: state.Definition.EquipmentPath,
AlarmName: state.Definition.AlarmName,
Kind: state.Definition.Kind,
Severity: state.Definition.Severity,
Message: message,
Condition: condition,
Emission: kind,
TimestampUtc: _clock(),
// Operator comment rides along on comment-bearing transitions — the condition
// state already carries it. Engine-driven transitions (Activated/Cleared/Shelved/…)
// and shelve ops (no comment param) leave it null.
Comment: kind switch
{
EmissionKind.Acknowledged => condition.LastAckComment,
EmissionKind.Confirmed => condition.LastConfirmComment,
EmissionKind.CommentAdded => condition.Comments.Count == 0 ? null : condition.Comments[^1].Text,
_ => null,
},
// Carry the per-alarm durable-historization opt-out through to subscribers. The historian
// adapter honors it to suppress ONLY the durable sink write; the live alerts fan-out is
// unaffected (it is not gated on this flag).
HistorizeToAveva: state.Definition.HistorizeToAveva,
// The worst input quality at evaluation time rides the transition so the projected
// full snapshot keeps quality consistent (no clobber-to-Good).
WorstInputStatusCode: worstInputStatus);
}
///
/// Build a standalone event carrying the new
/// worst-of-input quality. Emitted when the quality bucket moved but no Part 9 transition fired
/// (a Bad input freezes the condition; a Suppressed/None transition also leaves state unchanged).
/// The host routes it to the dedicated WriteAlarmQuality node path (annotate quality only,
/// no /alerts row, no historian write); the fan-out skips it.
///
private ScriptedAlarmEvent BuildQualityEmission(
AlarmState state, AlarmConditionState condition, uint worstInputStatus)
=> new(
AlarmId: state.Definition.AlarmId,
EquipmentPath: state.Definition.EquipmentPath,
AlarmName: state.Definition.AlarmName,
Kind: state.Definition.Kind,
Severity: state.Definition.Severity,
Message: MessageTemplate.Resolve(state.Definition.MessageTemplate, TryLookup),
Condition: condition,
Emission: EmissionKind.QualityChanged,
TimestampUtc: _clock(),
HistorizeToAveva: state.Definition.HistorizeToAveva,
WorstInputStatusCode: worstInputStatus);
/// Worst OPC UA StatusCode across a refilled read cache — the entry with the highest severity
/// bits (top 2). An input that has never been pushed is NOT a quality signal: it means "no data", which
/// the guard already handles by holding the condition. Counting it would
/// flash every scripted condition Bad at deploy until the first push and flood the quality path with
/// load-time annotations, so it is skipped (contributes Good). Empty / all-unpushed cache ⇒ Good (0).
/// "Never pushed" is a specific status code, not "the value is null"
/// (Finding 3). This used to skip every entry whose value was null, which made a genuine runtime Bad
/// indistinguishable from cold start — and that is exactly the shape a comms-lost driver produces, so
/// the case the skip was hiding was the case that matters most. The upstream source now stamps a
/// never-pushed path and nothing else does, so the two are tellable
/// apart. A null snapshot is still skipped: that is a cache the refill did not populate, not an
/// observation.
/// The comparison itself lives in
/// — the same rule the virtual-tag and Calculation seams now use, so "worst input wins, carrying its
/// sub-code, first-seen on a severity tie" is stated once. What stays here is the alarm-specific
/// filter: this engine's two skips are deliberately NOT part of the shared rule, because the
/// other two seams treat a never-pushed input as Bad rather than as absent.
private static uint WorstInputStatus(IReadOnlyDictionary cache)
{
// No entry at all — nothing was observed, so there is nothing to report on.
// Observed as "no data has ever arrived" — the cold-start placeholder, not a fault.
var observed = cache.Values.Where(v => v is not null && v.StatusCode != StatusNeverPushed);
return InputQuality.WorstOf(observed);
}
/// Update the tracked worst-quality bucket for an alarm; return true iff the 3-state bucket
/// (0 = Good, 1 = Uncertain, 2 = Bad) changed from the last observed value. Only called under
/// _evalGate.
private bool TrackQualityBucket(string alarmId, uint worstStatus)
{
var bucket = QualityBucket(worstStatus);
var prior = _lastQualityBucketByAlarmId.TryGetValue(alarmId, out var b) ? b : 0; // default Good
_lastQualityBucketByAlarmId[alarmId] = bucket;
return bucket != prior;
}
/// Collapse an OPC UA StatusCode's 2 severity bits (00/01/10/11) to a 3-state quality bucket
/// (0 = Good, 1 = Uncertain, 2 = Bad).
private static int QualityBucket(uint statusCode)
{
var severity = statusCode >> 30;
return severity >= 2 ? 2 : (int)severity;
}
/// A canonical worst StatusCode for an alarm's last-observed quality bucket, used by
/// the operator-command + shelving-timer emission paths (which don't re-read inputs) so an ack /
/// shelve while an input is Bad still carries Bad rather than resetting the condition to Good.
private uint LastWorstStatus(string alarmId)
=> (_lastQualityBucketByAlarmId.TryGetValue(alarmId, out var bucket) ? bucket : 0) switch
{
2 => 0x80000000u, // Bad
1 => 0x40000000u, // Uncertain
_ => 0u, // Good
};
///
/// Invoke the handler for a built emission. Must be
/// called OUTSIDE _evalGate: a slow subscriber would otherwise
/// block the gate for every other engine operation, and a subscriber
/// that re-enters the engine (e.g. calls AcknowledgeAsync) would
/// deadlock against the non-reentrant SemaphoreSlim.
///
private void FireEvent(ScriptedAlarmEvent evt)
{
try { OnEvent?.Invoke(this, evt); }
catch (Exception ex)
{
_engineLogger.Warning(ex, "ScriptedAlarmEngine OnEvent subscriber threw for {AlarmId}", evt.AlarmId);
}
}
private DataValueSnapshot? TryLookup(string path)
=> _valueCache.TryGetValue(path, out var v) ? v : null;
private void RunShelvingCheck()
{
if (_disposed) return;
var ids = _alarms.Keys.ToArray();
TrackBackgroundTask(() => ShelvingCheckAsync(ids, _shutdownToken));
}
///
/// Start a fire-and-forget task and register it so can await it.
///
///
/// The factory shape is the point: registration happens BEFORE the work starts.
/// Taking a meant the caller evaluated the argument — which starts the
/// work — and only then called in to register it. snapshots
/// _inFlight under _inFlightLock and synchronously awaits the snapshot, so a
/// Dispose landing in that start→Add window drained an empty (or short) set and returned
/// while a genuinely running SaveAsync was still in flight — exactly the outliving
/// store the drain exists to prevent. The window is removed, not shrunk: a placeholder is
/// registered under the lock first, and only then is the factory invoked, so any task that
/// runs at all is already visible to every subsequent snapshot.
/// The factory is invoked OUTSIDE _inFlightLock — it runs engine code
/// synchronously up to its first real await (and, via the injected clock, can re-enter
/// and this method), none of which belongs under that lock.
/// Post-disposal the call DECLINES: the factory is never invoked and nothing is
/// registered (the method returns silently — no throw, since every caller is a
/// fire-and-forget path that already treats a disposing engine as "stop"). The drain-latch
/// handover is the only caller that would otherwise leave _drainScheduled stuck true;
/// that is harmless, because a disposed engine drops pushes in
/// and so never consults the latch again.
///
/// Starts the background work. Invoked at most once, outside the lock.
private void TrackBackgroundTask(Func factory)
{
// RunContinuationsAsynchronously: the placeholder is completed from the real
// task's ExecuteSynchronously continuation, i.e. on the completing thread —
// this keeps an awaiter of the placeholder from being resumed inline there.
var placeholder = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
lock (_inFlightLock)
{
if (_disposed) return;
_inFlight.Add(placeholder.Task);
}
Task work;
try
{
// The null coalesce is defensive only: no current caller can return a null Task
// (every factory is an async lambda or returns a started Task). It costs nothing and
// keeps a future non-async factory from NREing the continuation wiring below.
work = factory() ?? Task.CompletedTask;
}
catch (Exception ex)
{
// A factory that throws synchronously started nothing, so the placeholder must
// not be left pending — a Dispose that already snapshotted it would block on it
// forever. Fault it (and observe the fault, so an un-snapshotted placeholder is
// not reported as an unobserved task exception), then let the caller see the throw
// it would have seen when the argument was evaluated at the call site.
lock (_inFlightLock) { _inFlight.Remove(placeholder.Task); }
placeholder.TrySetException(ex);
_ = placeholder.Task.Exception;
throw;
}
// ExecuteSynchronously so the removal runs on the completing thread — avoids
// scheduler delay between completion and unregistration that would otherwise
// let Dispose see a stale set. The placeholder mirrors the real task's outcome
// so Dispose's WhenAll still observes (and logs) a background failure.
work.ContinueWith(t =>
{
// Removing BEFORE completing the placeholder is safe (not a window where Dispose could
// miss still-running work): this continuation only runs once the REAL task has already
// finished, so anything a concurrent Dispose could still be waiting for is done.
lock (_inFlightLock) { _inFlight.Remove(placeholder.Task); }
if (t.IsFaulted)
{
placeholder.TrySetException(t.Exception!.InnerExceptions);
// Observe the mirror. A Dispose that snapshotted it still sees the fault
// through WhenAll (observing does not suppress propagation); one that did
// not would otherwise leave a second unobserved task exception behind.
_ = placeholder.Task.Exception;
}
else if (t.IsCanceled) placeholder.TrySetCanceled();
else placeholder.TrySetResult();
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
///
/// Test hook — triggers a shelving check synchronously without waiting for
/// the 5-second timer. Allows tests that inject a controllable clock to advance
/// time and immediately drive timed-shelve expiry.
///
internal void RunShelvingCheckForTest() => RunShelvingCheck();
///
/// Test hook — launches background work through the same registration path the
/// engine's own fire-and-forget callers use, so tests can pin the
/// register-before-start ordering directly.
///
/// Starts the background work.
internal void TrackBackgroundTaskForTest(Func factory) => TrackBackgroundTask(factory);
/// Test hook — the number of background tasks currently registered for the Dispose drain.
internal int InFlightCountForTest
{
get { lock (_inFlightLock) { return _inFlight.Count; } }
}
private async Task ShelvingCheckAsync(IReadOnlyList alarmIds, CancellationToken ct)
{
var pending = new List(0);
try
{
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Re-check after acquiring the gate: Timer.Dispose() does not wait for
// running callbacks, so a shelving-check callback that passed the _disposed
// check in RunShelvingCheck can arrive here after Dispose() has returned.
// Mutating _alarms or saving to a disposed store here is unsafe.
if (_disposed) return;
var now = _clock();
foreach (var id in alarmIds)
{
if (!_alarms.TryGetValue(id, out var state)) continue;
var result = Part9StateMachine.ApplyShelvingCheck(state.Condition, now);
if (!ReferenceEquals(result.State, state.Condition))
{
// Persist before updating in-memory so a store failure leaves
// both sides at the prior state.
await _store.SaveAsync(result.State, ct).ConfigureAwait(false);
_alarms[id] = state with { Condition = result.State };
if (result.Emission != EmissionKind.None)
{
var evt = BuildEmission(state, result.State, result.Emission, LastWorstStatus(id));
if (evt is not null) pending.Add(evt);
}
}
}
}
finally { _evalGate.Release(); }
}
catch (OperationCanceledException) when (_shutdownToken.IsCancellationRequested)
{
return; // Dispose cut the check short after its bound
}
catch (Exception ex)
{
_engineLogger.Warning(ex, "ScriptedAlarmEngine shelving-check failed");
return;
}
// Fire emissions OUTSIDE _evalGate.
foreach (var evt in pending) FireEvent(evt);
}
private void UnsubscribeFromUpstream()
{
foreach (var s in _upstreamSubscriptions)
{
try { s.Dispose(); } catch { }
}
_upstreamSubscriptions.Clear();
}
private void EnsureLoaded()
{
if (!_loaded) throw new InvalidOperationException(
"ScriptedAlarmEngine not loaded. Call LoadAsync first.");
}
/// Disposes the engine, cleaning up resources and waiting for in-flight background tasks.
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_shelvingTimer?.Dispose();
UnsubscribeFromUpstream();
// Drain any fire-and-forget background work (DrainDirtyAsync from
// OnUpstreamChange + ShelvingCheckAsync from the 5s timer) that started
// before _disposed = true was visible. The drain loop observes _disposed
// at the top of every iteration and exits, so it cannot spin here. Without
// this, a SaveAsync in
// flight can outlive the engine and write to a (possibly disposed) store
// after Dispose() has returned. The tasks re-check _disposed after
// acquiring the gate and bail out, but the await still has to complete.
// BOUNDED (R05:3): the drained tasks end in IAlarmStateStore.SaveAsync, and a stalled store
// (connection stall, lock, replication hiccup) used to hang this — and the host shutdown that
// calls it — indefinitely. Wait DisposeDrainTimeout; past it, cancel the tasks' shared token (a
// store that honours cancellation returns promptly), log, and give them one short grace period.
Task[] toAwait;
lock (_inFlightLock) { toAwait = [.. _inFlight]; }
if (toAwait.Length > 0)
{
var all = Task.WhenAll(toAwait);
try { all.WaitAsync(DisposeDrainTimeout).GetAwaiter().GetResult(); }
catch (TimeoutException)
{
_engineLogger.Warning(
"ScriptedAlarmEngine dispose: {Count} background task(s) still running after {Timeout} — cancelling them (a store call is stalled)",
toAwait.Length, DisposeDrainTimeout);
_shutdownCts.Cancel();
try { all.WaitAsync(TimeSpan.FromSeconds(1)).GetAwaiter().GetResult(); }
catch (Exception) { /* cancelled, faulted, or still stuck — we are leaving either way */ }
}
catch (Exception ex)
{
// Background task failures already logged inside ReevaluateAsync /
// ShelvingCheckAsync; surface here at debug so a parent shutdown is
// not noisy. The key invariant is that the tasks have COMPLETED.
_engineLogger.Debug(ex, "ScriptedAlarmEngine background task threw during shutdown drain");
}
}
// Safe to clear here: the Task.WhenAll drain above guaranteed no
// ReevaluateAsync / ShelvingCheckAsync is mid-flight, and _disposed=true
// prevents new background work from being queued (OnUpstreamChange bails on
// line 334). Previously the comment said "Do NOT clear _alarms",
// but that was when the engine called ScriptEvaluator.Compile directly and
// held the script ALCs through _alarms→AlarmState→TimedScriptEvaluator
// forever — leaving them rooted defeated the collectible-ALC unload.
// Clearing now drops the delegate references so the cache's Dispose call
// below can actually unload the emitted assemblies.
_alarms.Clear();
_alarmsReferencing.Clear();
_scratchByAlarmId.Clear();
_lastQualityBucketByAlarmId.Clear();
// Release the shutdown CTS once nothing can still be observing it through a registration: every
// drained task has completed. A task still stuck past the grace keeps its registration, so the CTS is
// left to the GC in that case (logged above).
if (toAwait.All(t => t.IsCompleted))
{
_shutdownCts.Dispose();
_shutdownCtsDisposed = true;
}
// Dispose every compiled-predicate ALC so the engine's shutdown actually
// releases the emitted assemblies. The drain above ensures no evaluator is
// mid-call; CompiledScriptCache.Dispose internally guards against use-after-
// dispose.
_compileCache.Dispose();
}
private sealed record AlarmState(
ScriptedAlarmDefinition Definition,
ScriptEvaluator? Evaluator, // null out of process: the worker holds the compiled form
IReadOnlySet Inputs,
IReadOnlyList TemplateTokens,
ILogger Logger,
AlarmConditionState Condition);
///
/// Per-alarm reusable evaluation scratch. The dictionary
/// is the same instance across every evaluation of the owning alarm — it is
/// cleared and refilled in on
/// each call. wraps that dictionary by reference, so a
/// refilled is what the predicate's
/// ctx.GetTag(path) calls observe.
///
///
/// Reuse is safe because serialises every
/// evaluation under _evalGate: two threads can never observe the same
/// scratch in a half-refilled state.
///
private sealed class AlarmScratch
{
/// Gets the read cache dictionary containing current upstream tag values.
public Dictionary ReadCache { get; }
/// Gets the predicate evaluation context.
public AlarmPredicateContext Context { get; }
/// The reason the last evaluation did not complete (timeout / refused / quarantined), or null
/// after a completed run — the once-per-change log gate for PredicateDidNotComplete.
public string? LastNonCompletion { get; set; }
///
/// Initializes a new AlarmScratch with the specified inputs, logger, and clock.
///
/// The set of input tag paths this alarm references.
/// The logger for this alarm's diagnostics.
/// Function providing the current UTC time.
public AlarmScratch(IReadOnlySet inputs, ILogger logger, Func clock)
{
// Pre-size to the expected input count so the first refill doesn't pay the
// dictionary-grow cost. The dictionary auto-grows if Inputs changes (it
// cannot under the current contract — Inputs is fixed at LoadAsync — but
// pre-sizing is defensive against future changes).
ReadCache = new Dictionary(inputs.Count, StringComparer.Ordinal);
// Context holds the read cache by reference. Refilling the dictionary
// updates what the context (and the script) observes.
Context = new AlarmPredicateContext(ReadCache, logger, clock);
}
}
}
///
/// One alarm emission the engine pushed to subscribers. Carries everything
/// downstream consumers (OPC UA alarm-source adapter + historian sink) need to
/// publish the event without re-querying the engine.
///
public sealed record ScriptedAlarmEvent(
string AlarmId,
string EquipmentPath,
string AlarmName,
AlarmKind Kind,
AlarmSeverity Severity,
string Message,
AlarmConditionState Condition,
EmissionKind Emission,
DateTime TimestampUtc,
string? Comment = null,
bool HistorizeToAveva = true,
// The worst OPC UA StatusCode across the alarm's input tags at evaluation time. A raw uint,
// since Core.ScriptedAlarms does not reference Commons. The host carries it to the condition
// node UNCHANGED — it used to be collapsed to a 3-state bucket and re-expanded, losing the sub-code.
// Default 0u == Good keeps every existing constructor call unchanged.
uint WorstInputStatusCode = 0u);
///
/// A loaded alarm's current condition, ready to be projected onto its OPC UA node.
/// Produced by on demand; it is not an
/// emission and never travels the path, so it can never
/// become an alerts row or a historian record. Deliberately a separate type from
/// — it carries no , so there is nothing
/// for a future refactor to mistake for a transition.
///
/// The alarm identifier (also its OPC UA condition NodeId).
/// The definition's severity bucket.
/// The message template resolved against the current value cache.
/// The Part 9 condition state the engine currently holds.
/// The last-observed worst OPC UA StatusCode across the alarm's inputs.
public sealed record ScriptedAlarmProjection(
string AlarmId,
AlarmSeverity Severity,
string Message,
AlarmConditionState Condition,
uint WorstInputStatusCode);
///
/// Upstream source abstraction — intentionally identical shape to the virtual-tag
/// engine's so Stream G can compose them behind one driver bridge.
///
public interface ITagUpstreamSource
{
/// Reads a tag value synchronously.
/// The tag path to read.
/// A data value snapshot containing the tag value and status.
DataValueSnapshot ReadTag(string path);
/// Subscribes to upstream tag changes.
/// The tag path to observe.
/// Callback invoked when the tag value changes.
/// A subscription handle that removes the observer when disposed.
IDisposable SubscribeTag(string path, Action observer);
}