Joseph Doherty
1310 lines
73 KiB
C#
1310 lines
73 KiB
C#
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;
|
|
|
|
/// <summary>
|
|
/// 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 <see cref="Part9StateMachine"/>,
|
|
/// persists state via <see cref="IAlarmStateStore"/>, and emits the resulting events
|
|
/// through <see cref="ScriptedAlarmSource"/> (which wires into the existing
|
|
/// <c>IAlarmSource</c> fan-out).
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// 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).
|
|
/// </para>
|
|
/// <para>
|
|
/// 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.
|
|
/// </para>
|
|
/// <para>
|
|
/// <b>Upstream pushes coalesce through a dirty set.</b> 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. <see cref="DrainDirtyAsync"/> 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:
|
|
/// <c>RefillReadCache</c> always reads <see cref="_valueCache"/>, 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 <see cref="_evalGate"/> was busy".
|
|
/// </para>
|
|
/// <para>
|
|
/// <b><see cref="IAlarmStateStore.SaveAsync"/> deliberately stays INSIDE the gate, before
|
|
/// the in-memory update.</b> 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
|
|
/// <c>Store_save_failure_leaves_in_memory_state_unchanged</c>. 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.
|
|
/// </para>
|
|
/// </remarks>
|
|
public sealed class ScriptedAlarmEngine : IDisposable
|
|
{
|
|
/// <summary>
|
|
/// OPC UA Part 4 <c>BadWaitingForInitialData</c> — the status the upstream source stamps on a
|
|
/// path nothing has ever pushed to, and the marker <see cref="WorstInputStatus"/> uses to tell
|
|
/// genuine cold start apart from a driver reporting Bad.
|
|
/// <para>Inlined rather than taken from <c>Commons.OpcUa.OpcUaStatus</c> because this project
|
|
/// deliberately does not reference <c>Commons</c> — the engine is pure evaluation logic over
|
|
/// <c>Core.Abstractions</c>. The value is fixed by the OPC UA spec, so the duplication cannot
|
|
/// drift; <c>DependencyMuxTagUpstreamSourceTests</c> pins the two agreeing.</para>
|
|
/// </summary>
|
|
private const uint StatusNeverPushed = 0x80320000u;
|
|
|
|
private readonly ITagUpstreamSource _upstream;
|
|
private readonly IAlarmStateStore _store;
|
|
private readonly ScriptLoggerFactory _loggerFactory;
|
|
private readonly ILogger _engineLogger;
|
|
private readonly Func<DateTime> _clock;
|
|
private readonly TimeSpan _scriptTimeout;
|
|
|
|
// 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<string, AlarmState> _alarms = new(StringComparer.Ordinal);
|
|
|
|
/// <summary>
|
|
/// Per-alarm reusable evaluation scratch. The read-cache dictionary and the
|
|
/// <see cref="AlarmPredicateContext"/> 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 <see cref="EvaluatePredicateToStateAsync"/> only
|
|
/// runs under <see cref="_evalGate"/>, which serialises every evaluation:
|
|
/// two threads can never observe the same scratch in a half-refilled state.
|
|
/// Cleared in <see cref="LoadAsync"/> alongside <see cref="_alarms"/>.
|
|
/// </summary>
|
|
private readonly ConcurrentDictionary<string, AlarmScratch> _scratchByAlarmId =
|
|
new(StringComparer.Ordinal);
|
|
|
|
/// <summary>
|
|
/// #478 — 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
|
|
/// <see cref="EmissionKind.QualityChanged"/> 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 <c>_evalGate</c>; cleared alongside <see cref="_alarms"/> on load/dispose.
|
|
/// </summary>
|
|
private readonly ConcurrentDictionary<string, int> _lastQualityBucketByAlarmId =
|
|
new(StringComparer.Ordinal);
|
|
|
|
/// <summary>
|
|
/// Compile cache for every alarm predicate. Routes <see cref="LoadAsync"/>'s
|
|
/// <see cref="ScriptEvaluator{TContext, TResult}.Compile"/> calls through the
|
|
/// cache so the collectible <see cref="System.Runtime.Loader.AssemblyLoadContext"/>
|
|
/// each compile produces is actually disposed on the publish-replace path:
|
|
/// the cache's <see cref="CompiledScriptCache{TContext, TResult}.Clear"/>
|
|
/// 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 <c>ScriptEvaluator.Compile</c> directly, which left
|
|
/// the ALCs rooted until the process exited — defeating -008 on the real path.
|
|
/// </summary>
|
|
private readonly CompiledScriptCache<AlarmPredicateContext, bool> _compileCache = new();
|
|
|
|
/// <summary>
|
|
/// 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).
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <b>Synchronization:</b> the returned <see cref="IReadOnlyDictionary{TKey, TValue}"/>
|
|
/// is the engine's live mutable read-cache. It is refilled in place by
|
|
/// <c>RefillReadCache</c> on every predicate evaluation, under <c>_evalGate</c>.
|
|
/// 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.
|
|
/// </remarks>
|
|
/// <param name="alarmId">The alarm identifier to look up.</param>
|
|
/// <returns>The alarm's read-cache dictionary if one has been allocated; otherwise null.</returns>
|
|
internal IReadOnlyDictionary<string, DataValueSnapshot>? TryGetScratchReadCacheForTest(string alarmId)
|
|
=> _scratchByAlarmId.TryGetValue(alarmId, out var s) ? s.ReadCache : null;
|
|
|
|
/// <summary>
|
|
/// Test-only diagnostic: returns the per-alarm <see cref="AlarmPredicateContext"/>
|
|
/// if one has been allocated, else null. Companion to
|
|
/// <see cref="TryGetScratchReadCacheForTest"/>.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <b>Synchronization:</b> the returned context wraps the same live
|
|
/// read-cache as <see cref="TryGetScratchReadCacheForTest"/> — the same
|
|
/// "don't iterate during an in-flight evaluation" caveat applies. Safe
|
|
/// for reference-identity assertions on a quiesced engine.
|
|
/// </remarks>
|
|
/// <param name="alarmId">The alarm identifier to look up.</param>
|
|
/// <returns>The alarm's predicate context if one has been allocated; otherwise null.</returns>
|
|
internal AlarmPredicateContext? TryGetScratchContextForTest(string alarmId)
|
|
=> _scratchByAlarmId.TryGetValue(alarmId, out var s) ? s.Context : null;
|
|
private readonly ConcurrentDictionary<string, DataValueSnapshot> _valueCache
|
|
= new(StringComparer.Ordinal);
|
|
private readonly Dictionary<string, HashSet<string>> _alarmsReferencing
|
|
= new(StringComparer.Ordinal); // tag path -> alarm ids
|
|
|
|
private readonly List<IDisposable> _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<Task> _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<string> _dirtyAlarmIds = new(StringComparer.Ordinal);
|
|
private readonly object _dirtyLock = new();
|
|
private bool _drainScheduled; // guarded by _dirtyLock
|
|
|
|
/// <summary>
|
|
/// Initializes a new ScriptedAlarmEngine with the provided dependencies.
|
|
/// </summary>
|
|
/// <param name="upstream">The upstream tag source for reading tag values.</param>
|
|
/// <param name="store">The alarm state store for persistence.</param>
|
|
/// <param name="loggerFactory">The factory for creating alarm loggers.</param>
|
|
/// <param name="engineLogger">The logger for engine-level diagnostics.</param>
|
|
/// <param name="clock">Optional function providing the current UTC time; defaults to DateTime.UtcNow.</param>
|
|
/// <param name="scriptTimeout">Optional timeout for script execution; defaults to the evaluator's default timeout.</param>
|
|
public ScriptedAlarmEngine(
|
|
ITagUpstreamSource upstream,
|
|
IAlarmStateStore store,
|
|
ScriptLoggerFactory loggerFactory,
|
|
ILogger engineLogger,
|
|
Func<DateTime>? clock = null,
|
|
TimeSpan? scriptTimeout = 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 ?? TimedScriptEvaluator<AlarmPredicateContext, bool>.DefaultTimeout;
|
|
}
|
|
|
|
/// <summary>Raised for every emission the Part9StateMachine produces that the engine should publish.</summary>
|
|
public event EventHandler<ScriptedAlarmEvent>? OnEvent;
|
|
|
|
/// <summary>Gets the collection of loaded alarm identifiers.</summary>
|
|
public IReadOnlyCollection<string> LoadedAlarmIds => _alarms.Keys.ToArray();
|
|
|
|
/// <summary>
|
|
/// Load a batch of alarm definitions. Compiles every predicate, aggregates any
|
|
/// compile failures into one <see cref="InvalidOperationException"/>, 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).
|
|
/// </summary>
|
|
/// <param name="definitions">The alarm definitions to load.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once every definition has been loaded and state restored.</returns>
|
|
public async Task LoadAsync(IReadOnlyList<ScriptedAlarmDefinition> definitions, CancellationToken ct)
|
|
{
|
|
if (_disposed) throw new ObjectDisposedException(nameof(ScriptedAlarmEngine));
|
|
if (definitions is null) throw new ArgumentNullException(nameof(definitions));
|
|
|
|
var pending = new List<ScriptedAlarmEvent>(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();
|
|
// 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();
|
|
|
|
var compileFailures = new List<string>();
|
|
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.
|
|
var evaluator = _compileCache.GetOrCompile(def.PredicateScriptSource);
|
|
var timed = new TimedScriptEvaluator<AlarmPredicateContext, bool>(evaluator, _scriptTimeout);
|
|
var logger = _loggerFactory.Create(def.AlarmId);
|
|
|
|
var templateTokens = MessageTemplate.ExtractTokenPaths(def.MessageTemplate);
|
|
var allInputs = new HashSet<string>(extraction.Reads, StringComparer.Ordinal);
|
|
foreach (var t in templateTokens) allInputs.Add(t);
|
|
|
|
_alarms[def.AlarmId] = new AlarmState(def, timed, 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<string>(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.
|
|
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);
|
|
}
|
|
|
|
_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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Current persisted state for <paramref name="alarmId"/>. Returns null for
|
|
/// unknown alarm. Mainly used for diagnostics + the Admin UI status page.
|
|
/// </summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <returns>The alarm's current condition state, or null if the alarm id is unknown.</returns>
|
|
public AlarmConditionState? GetState(string alarmId)
|
|
=> _alarms.TryGetValue(alarmId, out var s) ? s.Condition : null;
|
|
|
|
/// <summary>Gets the current persisted state for all loaded alarms.</summary>
|
|
/// <returns>The current condition state of every loaded alarm.</returns>
|
|
public IReadOnlyCollection<AlarmConditionState> GetAllStates()
|
|
=> _alarms.Values.Select(a => a.Condition).ToArray();
|
|
|
|
/// <summary>
|
|
/// #487 — 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 <b>project</b> the alarm onto an OPC UA condition node without waiting for
|
|
/// the next transition.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// <b>This is a read, not an emission.</b> It deliberately returns a distinct type rather
|
|
/// than a <see cref="ScriptedAlarmEvent"/>, and it never touches <see cref="OnEvent"/>. A
|
|
/// still-active alarm produces <see cref="EmissionKind.None"/> 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 / <c>alerts</c> row on every deploy, so the shape here is
|
|
/// intentionally one the emission machinery cannot consume.
|
|
/// </para>
|
|
/// <para>
|
|
/// <b>Synchronization.</b> Reads <c>_alarms</c> and <c>_valueCache</c> — both
|
|
/// <see cref="ConcurrentDictionary{TKey, TValue}"/> — without taking <c>_evalGate</c>, the
|
|
/// same posture as <see cref="GetState"/> / <see cref="GetAllStates"/>. 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.
|
|
/// </para>
|
|
/// </remarks>
|
|
/// <returns>One projection per loaded alarm; empty before the first <see cref="LoadAsync"/>.</returns>
|
|
public IReadOnlyList<ScriptedAlarmProjection> GetProjections()
|
|
{
|
|
var projections = new List<ScriptedAlarmProjection>(_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;
|
|
}
|
|
|
|
/// <summary>Acknowledges the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the acknowledgment.</param>
|
|
/// <param name="comment">An optional comment to attach to the acknowledgment.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the acknowledgment has been applied and persisted.</returns>
|
|
public Task AcknowledgeAsync(string alarmId, string user, string? comment, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyAcknowledge(cur, user, comment, _clock()));
|
|
|
|
/// <summary>Confirms the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the confirmation.</param>
|
|
/// <param name="comment">An optional comment to attach to the confirmation.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the confirmation has been applied and persisted.</returns>
|
|
public Task ConfirmAsync(string alarmId, string user, string? comment, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyConfirm(cur, user, comment, _clock()));
|
|
|
|
/// <summary>Applies a one-shot shelve to the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the shelve operation.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the shelve has been applied and persisted.</returns>
|
|
public Task OneShotShelveAsync(string alarmId, string user, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyOneShotShelve(cur, user, _clock()));
|
|
|
|
/// <summary>Applies a timed shelve to the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the shelve operation.</param>
|
|
/// <param name="unshelveAtUtc">The UTC time at which the shelve will automatically expire.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the timed shelve has been applied and persisted.</returns>
|
|
public Task TimedShelveAsync(string alarmId, string user, DateTime unshelveAtUtc, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyTimedShelve(cur, user, unshelveAtUtc, _clock()));
|
|
|
|
/// <summary>Removes any shelve from the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the unshelve operation.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the unshelve has been applied and persisted.</returns>
|
|
public Task UnshelveAsync(string alarmId, string user, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyUnshelve(cur, user, _clock()));
|
|
|
|
/// <summary>Enables the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the enable operation.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the enable has been applied and persisted.</returns>
|
|
public Task EnableAsync(string alarmId, string user, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyEnable(cur, user, _clock()));
|
|
|
|
/// <summary>Disables the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user performing the disable operation.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the disable has been applied and persisted.</returns>
|
|
public Task DisableAsync(string alarmId, string user, CancellationToken ct)
|
|
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyDisable(cur, user, _clock()));
|
|
|
|
/// <summary>Adds a comment to the specified alarm on behalf of the given user.</summary>
|
|
/// <param name="alarmId">The alarm identifier.</param>
|
|
/// <param name="user">The user adding the comment.</param>
|
|
/// <param name="text">The comment text.</param>
|
|
/// <param name="ct">The cancellation token.</param>
|
|
/// <returns>A task that completes once the comment has been recorded and persisted.</returns>
|
|
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<AlarmConditionState, TransitionResult> 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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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 <see cref="Dispose"/> can drain it.
|
|
/// </summary>
|
|
/// <param name="path">The upstream tag path that changed.</param>
|
|
/// <param name="value">The new data value snapshot.</param>
|
|
internal void OnUpstreamChange(string path, DataValueSnapshot value)
|
|
{
|
|
_valueCache[path] = value;
|
|
if (_disposed) return; // don't queue new work against a disposing engine
|
|
if (!_alarmsReferencing.TryGetValue(path, out var alarmIds)) return;
|
|
|
|
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(CancellationToken.None));
|
|
}
|
|
|
|
/// <summary>
|
|
/// The single drainer for the upstream-push dirty set. Swaps the accumulated
|
|
/// alarm ids out under <c>_dirtyLock</c> and re-evaluates that batch, looping
|
|
/// until the set is empty — so every push that landed while the previous pass
|
|
/// held <c>_evalGate</c> is served by the next pass rather than by a queue of
|
|
/// redundant per-push evaluations. Exactly one instance runs at a time
|
|
/// (<c>_drainScheduled</c>), 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".
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <b>INVARIANT: <c>_drainScheduled</c> is released on EVERY exit path, the
|
|
/// exceptional one included.</b> The flag is the only thing that lets
|
|
/// <see cref="OnUpstreamChange"/> start a drainer, so an escape that leaves it
|
|
/// stuck <c>true</c> permanently and silently stops evaluation for every scripted
|
|
/// alarm in the process: pushes keep filling <c>_valueCache</c> and the dirty set,
|
|
/// no drainer is ever started again, no predicate ever runs again, and nothing is
|
|
/// logged. <see cref="ReevaluateAsync"/> catches everything today, so the throw
|
|
/// surface is currently empty — the <c>finally</c> below is what keeps the
|
|
/// invariant true when it stops being (an audit-mode Serilog sink rethrowing out
|
|
/// of that catch block's own <c>Error()</c> call is enough).
|
|
/// </remarks>
|
|
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<string> alarmIds, CancellationToken ct)
|
|
{
|
|
var pending = new List<ScriptedAlarmEvent>(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;
|
|
|
|
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 (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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Evaluate the predicate + apply the resulting state-machine transition.
|
|
/// Returns the new condition state. If the transition produces an emission,
|
|
/// appends it to <paramref name="pendingEmissions"/> so the caller can fire
|
|
/// them after releasing <c>_evalGate</c> — keeping subscriber callbacks
|
|
/// outside the gate.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Every caller (LoadAsync and ReevaluateAsync) owns a <c>pending</c> 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.
|
|
/// </remarks>
|
|
private async Task<AlarmConditionState> EvaluatePredicateToStateAsync(
|
|
AlarmState state, AlarmConditionState seed, DateTime nowUtc, CancellationToken ct,
|
|
List<ScriptedAlarmEvent> 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);
|
|
|
|
// #478 — 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);
|
|
var qualityBucketChanged = TrackQualityBucket(state.Definition.AlarmId, worstStatus);
|
|
|
|
// 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 (qualityBucketChanged)
|
|
pendingEmissions.Add(BuildQualityEmission(state, seed, worstStatus));
|
|
return seed;
|
|
}
|
|
|
|
var context = scratch.Context;
|
|
|
|
bool predicateTrue;
|
|
try
|
|
{
|
|
predicateTrue = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
throw;
|
|
}
|
|
catch (ScriptTimeoutException tex)
|
|
{
|
|
state.Logger.Warning("Alarm predicate timed out after {Timeout} — state unchanged", tex.Timeout);
|
|
return seed;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
state.Logger.Error(ex, "Alarm predicate threw — state unchanged");
|
|
return seed;
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Refill <paramref name="cache"/> in place from <c>_valueCache</c>, falling
|
|
/// back to a synchronous <c>ITagUpstreamSource.ReadTag</c> for paths whose
|
|
/// first upstream push hasn't arrived yet. The dictionary is cleared and
|
|
/// repopulated under <c>_evalGate</c> so no concurrent reader can observe
|
|
/// a partial state. Replaces the old <c>BuildReadCache</c> which allocated a
|
|
/// fresh dictionary every call.
|
|
/// </summary>
|
|
private void RefillReadCache(
|
|
Dictionary<string, DataValueSnapshot> cache, IReadOnlySet<string> inputs)
|
|
{
|
|
cache.Clear();
|
|
foreach (var p in inputs)
|
|
cache[p] = _valueCache.TryGetValue(p, out var v) ? v : _upstream.ReadTag(p);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns true when every entry in <paramref name="cache"/> has a non-null value
|
|
/// and a Good-quality <see cref="DataValueSnapshot.StatusCode"/>. A false here lets
|
|
/// callers short-circuit script evaluation — predicates that unconditionally cast
|
|
/// <c>ctx.GetTag(path).Value</c> to a numeric type would otherwise throw
|
|
/// <see cref="NullReferenceException"/> until the upstream subscription delivers
|
|
/// its first push.
|
|
/// </summary>
|
|
private static bool AreInputsReady(IReadOnlyDictionary<string, DataValueSnapshot> 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;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Build (but do not fire) the <see cref="ScriptedAlarmEvent"/> for a
|
|
/// transition. Returns null for kinds that should not be published
|
|
/// (<see cref="EmissionKind.Suppressed"/> and
|
|
/// <see cref="EmissionKind.None"/>). Pure construction — called under
|
|
/// <c>_evalGate</c> so the message-template resolution uses a coherent
|
|
/// value-cache snapshot. The actual <see cref="OnEvent"/> dispatch is
|
|
/// done by <see cref="FireEvent(ScriptedAlarmEvent)"/> AFTER the gate is
|
|
/// released.
|
|
/// </summary>
|
|
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,
|
|
// #478 — the worst input quality at evaluation time rides the transition so the projected
|
|
// full snapshot keeps quality consistent (no clobber-to-Good).
|
|
WorstInputStatusCode: worstInputStatus);
|
|
}
|
|
|
|
/// <summary>
|
|
/// #478 — build a standalone <see cref="EmissionKind.QualityChanged"/> 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 <c>WriteAlarmQuality</c> node path (annotate quality only,
|
|
/// no <c>/alerts</c> row, no historian write); the <see cref="IAlarmSource"/> fan-out skips it.
|
|
/// </summary>
|
|
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);
|
|
|
|
/// <summary>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 <see cref="AreInputsReady"/> 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).
|
|
/// <para><b>"Never pushed" is a specific status code, not "the value is null" (#481, closing #478
|
|
/// Finding 3).</b> 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 <see cref="StatusNeverPushed"/> 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.</para>
|
|
/// <para><b>The comparison itself lives in <see cref="InputQuality.WorstOf"/></b> (archreview R3-02)
|
|
/// — 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
|
|
/// <em>filter</em>: 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.</para></summary>
|
|
private static uint WorstInputStatus(IReadOnlyDictionary<string, DataValueSnapshot> 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);
|
|
}
|
|
|
|
/// <summary>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
|
|
/// <c>_evalGate</c>.</summary>
|
|
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;
|
|
}
|
|
|
|
/// <summary>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).</summary>
|
|
private static int QualityBucket(uint statusCode)
|
|
{
|
|
var severity = statusCode >> 30;
|
|
return severity >= 2 ? 2 : (int)severity;
|
|
}
|
|
|
|
/// <summary>#478 — 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.</summary>
|
|
private uint LastWorstStatus(string alarmId)
|
|
=> (_lastQualityBucketByAlarmId.TryGetValue(alarmId, out var bucket) ? bucket : 0) switch
|
|
{
|
|
2 => 0x80000000u, // Bad
|
|
1 => 0x40000000u, // Uncertain
|
|
_ => 0u, // Good
|
|
};
|
|
|
|
/// <summary>
|
|
/// Invoke the <see cref="OnEvent"/> handler for a built emission. Must be
|
|
/// called OUTSIDE <c>_evalGate</c>: 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.
|
|
/// </summary>
|
|
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, CancellationToken.None));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Start a fire-and-forget task and register it so <see cref="Dispose"/> can await it.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para><b>The factory shape is the point: registration happens BEFORE the work starts.</b>
|
|
/// Taking a <see cref="Task"/> meant the caller evaluated the argument — which starts the
|
|
/// work — and only then called in to register it. <see cref="Dispose"/> snapshots
|
|
/// <c>_inFlight</c> under <c>_inFlightLock</c> 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 <c>SaveAsync</c> 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.</para>
|
|
/// <para>The factory is invoked OUTSIDE <c>_inFlightLock</c> — it runs engine code
|
|
/// synchronously up to its first real await (and, via the injected clock, can re-enter
|
|
/// <see cref="OnUpstreamChange"/> and this method), none of which belongs under that lock.</para>
|
|
/// <para><b>Post-disposal the call DECLINES: the factory is never invoked and nothing is
|
|
/// registered</b> (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 <c>_drainScheduled</c> stuck true;
|
|
/// that is harmless, because a disposed engine drops pushes in
|
|
/// <see cref="OnUpstreamChange"/> and so never consults the latch again.</para>
|
|
/// </remarks>
|
|
/// <param name="factory">Starts the background work. Invoked at most once, outside the lock.</param>
|
|
private void TrackBackgroundTask(Func<Task> 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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
internal void RunShelvingCheckForTest() => RunShelvingCheck();
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
/// <param name="factory">Starts the background work.</param>
|
|
internal void TrackBackgroundTaskForTest(Func<Task> factory) => TrackBackgroundTask(factory);
|
|
|
|
/// <summary>Test hook — the number of background tasks currently registered for the Dispose drain.</summary>
|
|
internal int InFlightCountForTest
|
|
{
|
|
get { lock (_inFlightLock) { return _inFlight.Count; } }
|
|
}
|
|
|
|
private async Task ShelvingCheckAsync(IReadOnlyList<string> alarmIds, CancellationToken ct)
|
|
{
|
|
var pending = new List<ScriptedAlarmEvent>(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 (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.");
|
|
}
|
|
|
|
/// <summary>Disposes the engine, cleaning up resources and waiting for in-flight background tasks.</summary>
|
|
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.
|
|
Task[] toAwait;
|
|
lock (_inFlightLock) { toAwait = [.. _inFlight]; }
|
|
if (toAwait.Length > 0)
|
|
{
|
|
try { Task.WhenAll(toAwait).GetAwaiter().GetResult(); }
|
|
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();
|
|
// 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,
|
|
TimedScriptEvaluator<AlarmPredicateContext, bool> Evaluator,
|
|
IReadOnlySet<string> Inputs,
|
|
IReadOnlyList<string> TemplateTokens,
|
|
ILogger Logger,
|
|
AlarmConditionState Condition);
|
|
|
|
/// <summary>
|
|
/// Per-alarm reusable evaluation scratch. The <see cref="ReadCache"/> dictionary
|
|
/// is the same instance across every evaluation of the owning alarm — it is
|
|
/// cleared and refilled in <see cref="ScriptedAlarmEngine.RefillReadCache"/> on
|
|
/// each call. <see cref="Context"/> wraps that dictionary by reference, so a
|
|
/// refilled <see cref="ReadCache"/> is what the predicate's
|
|
/// <c>ctx.GetTag(path)</c> calls observe.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Reuse is safe because <see cref="ScriptedAlarmEngine"/> serialises every
|
|
/// evaluation under <c>_evalGate</c>: two threads can never observe the same
|
|
/// scratch in a half-refilled state.
|
|
/// </remarks>
|
|
private sealed class AlarmScratch
|
|
{
|
|
/// <summary>Gets the read cache dictionary containing current upstream tag values.</summary>
|
|
public Dictionary<string, DataValueSnapshot> ReadCache { get; }
|
|
/// <summary>Gets the predicate evaluation context.</summary>
|
|
public AlarmPredicateContext Context { get; }
|
|
|
|
/// <summary>
|
|
/// Initializes a new AlarmScratch with the specified inputs, logger, and clock.
|
|
/// </summary>
|
|
/// <param name="inputs">The set of input tag paths this alarm references.</param>
|
|
/// <param name="logger">The logger for this alarm's diagnostics.</param>
|
|
/// <param name="clock">Function providing the current UTC time.</param>
|
|
public AlarmScratch(IReadOnlySet<string> inputs, ILogger logger, Func<DateTime> 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<string, DataValueSnapshot>(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);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
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,
|
|
// #478 — the worst OPC UA StatusCode across the alarm's input tags at evaluation time. A raw uint,
|
|
// since Core.ScriptedAlarms does not reference Commons. Since #519 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);
|
|
|
|
/// <summary>
|
|
/// #487 — a loaded alarm's current condition, ready to be projected onto its OPC UA node.
|
|
/// Produced by <see cref="ScriptedAlarmEngine.GetProjections"/> on demand; it is <b>not</b> an
|
|
/// emission and never travels the <see cref="ScriptedAlarmEngine.OnEvent"/> path, so it can never
|
|
/// become an <c>alerts</c> row or a historian record. Deliberately a separate type from
|
|
/// <see cref="ScriptedAlarmEvent"/> — it carries no <see cref="EmissionKind"/>, so there is nothing
|
|
/// for a future refactor to mistake for a transition.
|
|
/// </summary>
|
|
/// <param name="AlarmId">The alarm identifier (also its OPC UA condition NodeId).</param>
|
|
/// <param name="Severity">The definition's severity bucket.</param>
|
|
/// <param name="Message">The message template resolved against the current value cache.</param>
|
|
/// <param name="Condition">The Part 9 condition state the engine currently holds.</param>
|
|
/// <param name="WorstInputStatusCode">The last-observed worst OPC UA StatusCode across the alarm's inputs.</param>
|
|
public sealed record ScriptedAlarmProjection(
|
|
string AlarmId,
|
|
AlarmSeverity Severity,
|
|
string Message,
|
|
AlarmConditionState Condition,
|
|
uint WorstInputStatusCode);
|
|
|
|
/// <summary>
|
|
/// Upstream source abstraction — intentionally identical shape to the virtual-tag
|
|
/// engine's so Stream G can compose them behind one driver bridge.
|
|
/// </summary>
|
|
public interface ITagUpstreamSource
|
|
{
|
|
/// <summary>Reads a tag value synchronously.</summary>
|
|
/// <param name="path">The tag path to read.</param>
|
|
/// <returns>A data value snapshot containing the tag value and status.</returns>
|
|
DataValueSnapshot ReadTag(string path);
|
|
|
|
/// <summary>Subscribes to upstream tag changes.</summary>
|
|
/// <param name="path">The tag path to observe.</param>
|
|
/// <param name="observer">Callback invoked when the tag value changes.</param>
|
|
/// <returns>A subscription handle that removes the observer when disposed.</returns>
|
|
IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer);
|
|
}
|