fix(runtime): restart-safe alarm host, mux last-value replay, VT log rate limit, Calculation remediation (E4)
- R02:H1 ScriptedAlarmHostActor gains a factory Props (Func<(upstream, engine)>) so every incarnation builds and owns its own engine; a supervision restart no longer re-creates the actor around a disposed engine. The legacy instance-taking Props stays source-compatible and is now restart-safe in the weaker sense (PreRestart marks the stop as a restart, PostStop then detaches instead of disposing). DriverHostActor still spawns through the legacy overload (not owned by this lane — F3). - R02:H3 OnAlarmCommand runs every engine op under a 10 s token linked to the actor's lifetime (DefaultCommandTimeout / Props commandTimeout) and logs a Warning on timeout instead of suspending the mailbox on a stalled store. - R02:L3 DependencyMuxTagUpstreamSource.RetainOnly(loaded dep-ref union), called after every successful load. - R02:P6 VirtualTagActor's per-failure ScriptLogEntry is rate-limited per tag (ScriptLogRateLimiter, 20/min + one summary); the Bad publish and the metric are untouched. - R07:H6 DependencyMuxActor retains the last (value, status, ts) per ever-registered ref (through unregisters) and replays them in one batch to a RegisterInterest(ReplayRetained: true) — VirtualTagActor, the Calculation mux adapter and the alarm host opt in; the historization recorder does not (a replay there would record a stale sample per deploy). CalculationDriver records inputs regardless of _running so a replay landing before InitializeAsync still satisfies the change-gate. - R07:P4 CalculationEvaluator negative-caches compile failures by source hash (one Roslyn attempt + one Warning per source until redeploy); the driver's script-log entry fires on the Good→Bad transition or a change of reason, matching the once-per-transition Bad publish. - R07:H8 array results dedup structurally. - R07:L2 ShutdownAsync drops the compiled scripts (collectible ALCs). - R07:U5 DraftValidator rule CalculationNoEffectiveTrigger: a tag with no timer whose change trigger is off or whose script has no literal ctx.GetTag read can never evaluate. Refs #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -452,6 +452,54 @@ public static class DraftValidator
|
||||
"Calculation tags form a dependency cycle (members: " + string.Join(", ", cycle) +
|
||||
"); a calc→calc cycle is a live oscillation loop and must be broken before deploy",
|
||||
cycle.Count > 0 ? cycle[0] : null));
|
||||
|
||||
// Rule 3 — effective-trigger gate (R07:U5). A calc tag evaluates on a declared dependency
|
||||
// changing (change-trigger: the literal ctx.GetTag paths) and/or on a timer. A change-triggered
|
||||
// tag whose script has NO literal reads (`return ctx.Now.Hour;`, a dynamic-path read) and no timer
|
||||
// never evaluates at all: it sits BadWaitingForInitialData forever while the driver reports
|
||||
// Healthy, and nothing warned. Same for a tag that is neither change- nor timer-triggered.
|
||||
foreach (var (t, _, scriptId) in calcTags)
|
||||
{
|
||||
var (changeTriggered, timerIntervalMs) = ParseTriggers(t.TagConfig);
|
||||
if (timerIntervalMs is > 0) continue;
|
||||
var source = scriptId is not null && scriptSourceById.TryGetValue(scriptId, out var s) ? s : null;
|
||||
var depCount = EquipmentScriptPaths.ExtractDependencyRefs(source).Count;
|
||||
if (changeTriggered && depCount > 0) continue;
|
||||
errors.Add(new("CalculationNoEffectiveTrigger",
|
||||
changeTriggered
|
||||
? $"Calculation tag '{t.TagId}' is change-triggered but its script has no literal ctx.GetTag read, " +
|
||||
"and it has no timerIntervalMs — nothing would ever evaluate it. Add a dependency read or a timer."
|
||||
: $"Calculation tag '{t.TagId}' has changeTriggered=false and no timerIntervalMs — nothing would ever evaluate it. " +
|
||||
"Enable the change trigger or add a timer.",
|
||||
t.TagId));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Read the trigger fields from a calc tag's <c>TagConfig</c>: <c>changeTriggered</c>
|
||||
/// (default <c>true</c>, the driver's own default) and <c>timerIntervalMs</c> (null when absent or not a
|
||||
/// number). Never throws — malformed JSON yields the defaults.</summary>
|
||||
private static (bool ChangeTriggered, int? TimerIntervalMs) ParseTriggers(string? tagConfig)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(tagConfig)) return (true, null);
|
||||
try
|
||||
{
|
||||
using var doc = System.Text.Json.JsonDocument.Parse(tagConfig);
|
||||
var root = doc.RootElement;
|
||||
if (root.ValueKind != System.Text.Json.JsonValueKind.Object) return (true, null);
|
||||
var changeTriggered = true;
|
||||
if (root.TryGetProperty("changeTriggered", out var ct)
|
||||
&& ct.ValueKind is System.Text.Json.JsonValueKind.False or System.Text.Json.JsonValueKind.True)
|
||||
changeTriggered = ct.GetBoolean();
|
||||
int? timer = null;
|
||||
if (root.TryGetProperty("timerIntervalMs", out var tm)
|
||||
&& tm.ValueKind == System.Text.Json.JsonValueKind.Number && tm.TryGetInt32(out var ms))
|
||||
timer = ms;
|
||||
return (changeTriggered, timer);
|
||||
}
|
||||
catch (System.Text.Json.JsonException)
|
||||
{
|
||||
return (true, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Extract the <c>scriptId</c> string from a calc tag's schemaless <c>TagConfig</c> JSON. Never
|
||||
|
||||
@@ -338,6 +338,11 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
StopTimers();
|
||||
// The pump exits at its next iteration; an evaluation already in flight is bounded by RunTimeout.
|
||||
lock (_lock) _pending.Clear();
|
||||
// R07:L2 — nothing in the runtime disposes a driver, and every config change is a respawn, so
|
||||
// the compiled scripts' collectible ALCs survived each generation until GC got round to the whole
|
||||
// driver graph. Drop them here (idempotent; a later InitializeAsync recompiles on first use). An
|
||||
// in-flight run that races this is already handled by the evaluator's retry-once path.
|
||||
_evaluator.ClearCompiledScripts();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
@@ -363,11 +368,15 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
/// <inheritdoc />
|
||||
public void OnDependencyValue(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
|
||||
{
|
||||
if (!_running) return;
|
||||
lock (_lock)
|
||||
{
|
||||
// Recorded REGARDLESS of _running (R07:H6): the mux adapter registers interest at construction,
|
||||
// so inputs — the mux's retained-value replay in particular — can land before InitializeAsync.
|
||||
// Dropping them left every calc tag waiting for each input to change again; keeping them means
|
||||
// the first trigger after init evaluates from current inputs. Only the EVALUATION is gated.
|
||||
_depValues[rawPath] = new DataValueSnapshot(value, statusCode, timestampUtc, timestampUtc);
|
||||
_arrivedDeps.Add(rawPath);
|
||||
if (!_running) return;
|
||||
if (!_changeDependents.TryGetValue(rawPath, out var dependents) || dependents.Count == 0) return;
|
||||
|
||||
// QUEUE each change-triggered dependent — never evaluate here. This
|
||||
@@ -600,7 +609,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
var result = await _evaluator.EvaluateAsync(def.RawPath, def.ScriptSource, pending.Dependencies, _evalCts.Token)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
string? failureReason = null;
|
||||
string? failureToLog = null;
|
||||
(object? Value, uint StatusCode)? toPublish;
|
||||
lock (_lock)
|
||||
{
|
||||
@@ -610,16 +619,24 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
|
||||
if (!result.Success)
|
||||
{
|
||||
failureReason = result.Reason ?? "evaluator failure";
|
||||
var reason = result.Reason ?? "evaluator failure";
|
||||
toPublish = MarkEvaluationFailed(st, pending.TimestampUtc);
|
||||
// The script-log entry fires on the Good→Bad TRANSITION, or when the failure changes
|
||||
// (R07:P4) — not on every trigger of a tag that is already Bad for the same reason, which
|
||||
// flooded the script log at the input's cadence. The Bad publish was already once per
|
||||
// transition; the log now matches it.
|
||||
if (toPublish is not null || !string.Equals(st.LastFailureReason, reason, StringComparison.Ordinal))
|
||||
failureToLog = reason;
|
||||
st.LastFailureReason = reason;
|
||||
}
|
||||
else
|
||||
{
|
||||
st.LastFailureReason = null;
|
||||
toPublish = MarkEvaluationSucceeded(st, result.Value, pending.WorstInputStatus, pending.TimestampUtc);
|
||||
}
|
||||
}
|
||||
|
||||
if (failureReason is not null) PublishScriptLog(def, failureReason);
|
||||
if (failureToLog is not null) PublishScriptLog(def, failureToLog);
|
||||
if (toPublish is { } p) Publish(def.RawPath, p.Value, p.StatusCode, pending.TimestampUtc);
|
||||
}
|
||||
|
||||
@@ -638,7 +655,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
private (object? Value, uint StatusCode)? MarkEvaluationSucceeded(
|
||||
TagState st, object? value, uint statusCode, DateTime timestampUtc)
|
||||
{
|
||||
if (st.HasPublished && st.LastPublishedStatus == statusCode && st.HasValue && Equals(st.LastValue, value))
|
||||
if (st.HasPublished && st.LastPublishedStatus == statusCode && st.HasValue && ValuesEqual(st.LastValue, value))
|
||||
return null;
|
||||
|
||||
st.HasValue = true;
|
||||
@@ -650,6 +667,17 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
return (value, statusCode);
|
||||
}
|
||||
|
||||
/// <summary>Value equality for the dedup: structural for arrays (R07:H8 — <c>object.Equals</c> is reference
|
||||
/// equality for a <c>double[]</c>, so an array-returning timer script republished every tick through
|
||||
/// the dual-namespace fan-out and the historian), ordinary <see cref="object.Equals(object?, object?)"/>
|
||||
/// otherwise.</summary>
|
||||
private static bool ValuesEqual(object? a, object? b)
|
||||
{
|
||||
if (a is Array x && b is Array y)
|
||||
return System.Collections.StructuralComparisons.StructuralEqualityComparer.Equals(x, y);
|
||||
return Equals(a, b);
|
||||
}
|
||||
|
||||
/// <summary>Record that this tag's INPUTS are Bad, so no evaluation ran. Publishes the
|
||||
/// worst input's status — carrying the last known value — <b>once per change of that status</b>: a
|
||||
/// repeat of the same Bad is silent, while <c>BadNoCommunication</c> arriving over an earlier
|
||||
@@ -727,6 +755,12 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
/// <summary>Test/diagnostic accessor: number of timer ticks skipped for in-flight groups.</summary>
|
||||
internal long TimerSkippedTickCount => _timers?.SkippedTickCount ?? 0;
|
||||
|
||||
/// <summary>Test/diagnostic accessor: sources currently remembered as failing to compile.</summary>
|
||||
internal int CompileFailureCountForTest => _evaluator.CompileFailureCount;
|
||||
|
||||
/// <summary>Test/diagnostic accessor: compiled scripts currently held by the evaluator's cache.</summary>
|
||||
internal int CompiledScriptCountForTest => _evaluator.CompiledScriptCount;
|
||||
|
||||
/// <summary>Test/diagnostic accessor: <see langword="true"/> when no change-triggered evaluation is
|
||||
/// queued or in flight. A test that used to assert straight after <see cref="OnDependencyValue"/> —
|
||||
/// which evaluated inline — waits on this instead, so the assertion still runs against a settled
|
||||
@@ -770,6 +804,10 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
|
||||
/// now carries the failing INPUT's sub-code rather than a fixed <c>BadInternalError</c>.</summary>
|
||||
public uint LastPublishedStatus;
|
||||
public DateTime LastTimestampUtc;
|
||||
|
||||
/// <summary>The reason of the last failed evaluation (null after a success / inputs-Bad hold) — the
|
||||
/// gate that keeps the script-log entry to the transition and to a CHANGE of reason.</summary>
|
||||
public string? LastFailureReason;
|
||||
}
|
||||
|
||||
private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
@@ -58,6 +59,13 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO
|
||||
private readonly SerilogLogger _scriptRoot;
|
||||
private readonly TimeSpan _runTimeout;
|
||||
private readonly ScriptRunGuard _guard;
|
||||
|
||||
// Negative compile cache (R07:P4), keyed by source hash. A compile error is a property of the SOURCE,
|
||||
// which never changes between triggers (it is fixed in the deploy artifact) — yet every trigger of a
|
||||
// broken tag used to re-run a full Roslyn compile under the driver-wide evaluation gate, starving the
|
||||
// healthy tags behind it, and log the same Warning each time. The failure is now remembered until the
|
||||
// next ClearCompiledScripts (a redeploy), and the host Warning fires once per source.
|
||||
private readonly ConcurrentDictionary<string, string> _compileFailures = new(StringComparer.Ordinal);
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>Initializes a new <see cref="CalculationEvaluator"/>.</summary>
|
||||
@@ -180,6 +188,11 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO
|
||||
calcTagId, path),
|
||||
logger: scriptLog);
|
||||
|
||||
// A source that already failed to compile fails the same way without touching Roslyn (R07:P4).
|
||||
var sourceKey = HashSource(scriptSource);
|
||||
if (_compileFailures.TryGetValue(sourceKey, out var knownFailure))
|
||||
return VirtualTagEvalResult.Failure(knownFailure);
|
||||
|
||||
// Fetch + run inside a bounded loop so an ObjectDisposedException caused by a concurrent
|
||||
// ClearCompiledScripts (which disposed the evaluator between GetOrCompile and the run's
|
||||
// disposed guard) re-fetches and retries exactly once. The loop runs at most twice.
|
||||
@@ -192,13 +205,11 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO
|
||||
}
|
||||
catch (CompilationErrorException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Calculation {Id}: Roslyn compile failed", calcTagId);
|
||||
return VirtualTagEvalResult.Failure($"compile error: {ex.Message}");
|
||||
return RememberCompileFailure(sourceKey, calcTagId, $"compile error: {ex.Message}", ex, "Roslyn compile failed");
|
||||
}
|
||||
catch (ScriptSandboxViolationException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Calculation {Id}: sandbox violation", calcTagId);
|
||||
return VirtualTagEvalResult.Failure($"sandbox violation: {ex.Message}");
|
||||
return RememberCompileFailure(sourceKey, calcTagId, $"sandbox violation: {ex.Message}", ex, "sandbox violation");
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
@@ -258,10 +269,29 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Record a compile failure for <paramref name="sourceKey"/> and log it ONCE — the next trigger
|
||||
/// of the same source returns the remembered reason without a Roslyn round-trip or a second Warning.</summary>
|
||||
private VirtualTagEvalResult RememberCompileFailure(string sourceKey, string calcTagId, string reason, Exception ex, string what)
|
||||
{
|
||||
if (_compileFailures.TryAdd(sourceKey, reason))
|
||||
_logger.LogWarning(ex, "Calculation {Id}: {What} (remembered until the next redeploy)", calcTagId, what);
|
||||
return VirtualTagEvalResult.Failure(reason);
|
||||
}
|
||||
|
||||
private static string HashSource(string source)
|
||||
=> Convert.ToHexString(System.Security.Cryptography.SHA256.HashData(System.Text.Encoding.UTF8.GetBytes(source)));
|
||||
|
||||
/// <summary>Test/diagnostic: the number of sources currently remembered as failing to compile.</summary>
|
||||
public int CompileFailureCount => _compileFailures.Count;
|
||||
|
||||
/// <summary>Test/diagnostic: the number of compiled scripts currently cached.</summary>
|
||||
public int CompiledScriptCount => _cache.Count;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void ClearCompiledScripts()
|
||||
{
|
||||
_cache.Clear();
|
||||
_compileFailures.Clear();
|
||||
// The compile-cache clear is this evaluator's deploy boundary — a redeploy lifts every quarantine
|
||||
// (an edited source already lifts its own through the guard's source identity).
|
||||
_guard.ResetAll();
|
||||
@@ -279,5 +309,6 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
_cache.Dispose();
|
||||
_compileFailures.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
+26
-2
@@ -42,8 +42,10 @@ public sealed class DependencyMuxTagUpstreamSource : ITagUpstreamSource
|
||||
// publish.
|
||||
private const uint StatusNeverPushed = OpcUaStatus.BadWaitingForInitialData;
|
||||
|
||||
// Intentionally never pruned: cold-start semantics depend on retained last values so
|
||||
// ReadTag can answer synchronously for any path that has ever been pushed.
|
||||
// Retained last values: ReadTag answers synchronously for any path that has been pushed. Pruned
|
||||
// ONLY through RetainOnly — the host calls it with the loaded dependency-ref union after every
|
||||
// (re)load, so a renamed/removed raw path (a RawTree rename is a surgical remove+add) stops
|
||||
// costing a boxed value + two DateTimes for the process lifetime (R02:L3).
|
||||
private readonly ConcurrentDictionary<string, DataValueSnapshot> _cache
|
||||
= new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentDictionary<string, ImmutableList<Action<string, DataValueSnapshot>>> _observers
|
||||
@@ -91,6 +93,28 @@ public sealed class DependencyMuxTagUpstreamSource : ITagUpstreamSource
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Drop every cached snapshot whose path is not in <paramref name="paths"/>. Observers are
|
||||
/// untouched. Called by the host after each successful engine (re)load with the union of the loaded
|
||||
/// alarms' dependency refs — exactly the set the mux will push from now on — so paths no alarm
|
||||
/// references any more stop being retained.
|
||||
/// </summary>
|
||||
/// <param name="paths">The paths to keep.</param>
|
||||
/// <returns>The number of cached snapshots dropped.</returns>
|
||||
public int RetainOnly(IReadOnlySet<string> paths)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(paths);
|
||||
var dropped = 0;
|
||||
foreach (var key in _cache.Keys)
|
||||
{
|
||||
if (!paths.Contains(key) && _cache.TryRemove(key, out _)) dropped++;
|
||||
}
|
||||
return dropped;
|
||||
}
|
||||
|
||||
/// <summary>Test/diagnostic: the number of paths currently retained.</summary>
|
||||
public int RetainedPathCount => _cache.Count;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public DataValueSnapshot ReadTag(string path)
|
||||
{
|
||||
|
||||
@@ -23,10 +23,16 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
||||
/// cluster <c>alerts</c> DistributedPubSub topic.
|
||||
///
|
||||
/// <para>
|
||||
/// The host is the engine's lifecycle owner: the caller (the driver-role startup) builds
|
||||
/// the <see cref="DependencyMuxTagUpstreamSource"/>, constructs a <see cref="ScriptedAlarmEngine"/>
|
||||
/// <b>around that same upstream</b>, and passes both here. The host disposes the engine in
|
||||
/// <see cref="PostStop"/>.
|
||||
/// The host is the engine's lifecycle owner. The restart-safe shape (R02:H1) is the
|
||||
/// <b>factory</b> <see cref="Props(IActorRef, IActorRef?, Func{ValueTuple{DependencyMuxTagUpstreamSource, ScriptedAlarmEngine}}, NodeId?, Func{int}?, ITelemetryLocalHub?, TimeSpan?)"/>:
|
||||
/// every incarnation builds its own <see cref="DependencyMuxTagUpstreamSource"/> + a
|
||||
/// <see cref="ScriptedAlarmEngine"/> around it, and disposes them in <see cref="PostStop"/>.
|
||||
/// Akka's default restart runs <c>PostStop</c> then re-creates the actor from the SAME Props, so the
|
||||
/// legacy instance-taking Props — a fixed engine captured in the Props — restarted around a
|
||||
/// <em>disposed</em> engine and every later <c>LoadAsync</c> threw: the node's scripted alarms
|
||||
/// silently died. The legacy overload is kept source-compatible but is restart-safe only in the
|
||||
/// weaker sense that a restart now DETACHES rather than disposes its engine (see
|
||||
/// <see cref="PreRestart"/>); production should spawn through the factory.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
@@ -99,6 +105,9 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
/// Warning on every clean shutdown.</summary>
|
||||
private sealed record AlarmsLoadCanceled;
|
||||
|
||||
/// <summary>Default bound on one inbound alarm command's engine op (<see cref="OnAlarmCommand"/>): 10 s.</summary>
|
||||
public static readonly TimeSpan DefaultCommandTimeout = TimeSpan.FromSeconds(10);
|
||||
|
||||
private readonly IActorRef _publishActor;
|
||||
private readonly IActorRef? _mux;
|
||||
private readonly DependencyMuxTagUpstreamSource _upstream;
|
||||
@@ -107,6 +116,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private readonly EventHandler<ScriptedAlarmEvent> _onEngineEvent;
|
||||
private readonly TimeSpan _commandTimeout;
|
||||
|
||||
/// <summary>True when THIS incarnation built its engine (factory Props) and therefore disposes it on
|
||||
/// every stop, restart included. False for the legacy instance-taking Props: the engine outlives the
|
||||
/// incarnation, so a restart only detaches from it and a final stop disposes it.</summary>
|
||||
private readonly bool _ownsEngine;
|
||||
|
||||
/// <summary>Set by <see cref="PreRestart"/> so the <see cref="PostStop"/> it triggers can tell a restart
|
||||
/// (the next incarnation reuses a legacy engine) from a genuine stop.</summary>
|
||||
private bool _restarting;
|
||||
|
||||
/// <summary>Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
|
||||
/// snapshot (null = unknown until the first snapshot arrives, or no <see cref="_localNode"/> wired). The
|
||||
@@ -152,6 +171,8 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
/// <param name="telemetryHub">Optional node-local live-telemetry hub (Phase 5) each cluster-wide
|
||||
/// <c>alerts</c> transition is also emitted into; null (tests) skips the emit.</param>
|
||||
/// <returns>The <see cref="Akka.Actor.Props"/> used to instantiate the actor.</returns>
|
||||
/// <remarks>Legacy, instance-taking shape — see the class remarks: a supervision restart re-creates the
|
||||
/// actor around these SAME instances, so prefer the factory overload in production.</remarks>
|
||||
public static Props Props(
|
||||
IActorRef publishActor,
|
||||
IActorRef? mux,
|
||||
@@ -162,6 +183,60 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
ITelemetryLocalHub? telemetryHub = null) =>
|
||||
Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider, telemetryHub));
|
||||
|
||||
/// <summary>
|
||||
/// Restart-safe factory Props (R02:H1): each incarnation of the actor calls
|
||||
/// <paramref name="engineFactory"/> in its constructor and owns what it returns — the engine must
|
||||
/// be constructed <b>around</b> the returned upstream — so a supervision restart gets a fresh,
|
||||
/// undisposed engine and the next <see cref="ApplyScriptedAlarms"/> loads cleanly.
|
||||
/// </summary>
|
||||
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
|
||||
/// <param name="mux">Optional dependency multiplexer the host registers interest with.</param>
|
||||
/// <param name="engineFactory">Builds a fresh (upstream, engine) pair; called once per incarnation.</param>
|
||||
/// <param name="localNode">The local cluster node id for the Primary alerts-emit gate; null ⇒ default-emit.</param>
|
||||
/// <param name="driverMemberCountProvider">Test seam for the alerts-emit gate's driver-member count.</param>
|
||||
/// <param name="telemetryHub">Optional node-local live-telemetry hub.</param>
|
||||
/// <param name="commandTimeout">Bound on one inbound alarm command's engine op; null ⇒ <see cref="DefaultCommandTimeout"/>.</param>
|
||||
/// <returns>The <see cref="Akka.Actor.Props"/> used to instantiate the actor.</returns>
|
||||
public static Props Props(
|
||||
IActorRef publishActor,
|
||||
IActorRef? mux,
|
||||
Func<(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine)> engineFactory,
|
||||
NodeId? localNode = null,
|
||||
Func<int>? driverMemberCountProvider = null,
|
||||
ITelemetryLocalHub? telemetryHub = null,
|
||||
TimeSpan? commandTimeout = null) =>
|
||||
Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, engineFactory, localNode, driverMemberCountProvider, telemetryHub, commandTimeout));
|
||||
|
||||
/// <summary>Factory-built incarnation: owns (and on every stop disposes) the engine it builds.</summary>
|
||||
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
|
||||
/// <param name="mux">Optional dependency multiplexer the host registers interest with.</param>
|
||||
/// <param name="engineFactory">Builds this incarnation's (upstream, engine) pair.</param>
|
||||
/// <param name="localNode">The local cluster node id for the Primary alerts-emit gate.</param>
|
||||
/// <param name="driverMemberCountProvider">Test seam for the alerts-emit gate's driver-member count.</param>
|
||||
/// <param name="telemetryHub">Optional node-local live-telemetry hub.</param>
|
||||
/// <param name="commandTimeout">Bound on one inbound alarm command's engine op.</param>
|
||||
public ScriptedAlarmHostActor(
|
||||
IActorRef publishActor,
|
||||
IActorRef? mux,
|
||||
Func<(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine)> engineFactory,
|
||||
NodeId? localNode = null,
|
||||
Func<int>? driverMemberCountProvider = null,
|
||||
ITelemetryLocalHub? telemetryHub = null,
|
||||
TimeSpan? commandTimeout = null)
|
||||
: this(publishActor, mux, Build(engineFactory), localNode, driverMemberCountProvider, telemetryHub, ownsEngine: true, commandTimeout)
|
||||
{
|
||||
}
|
||||
|
||||
private static (DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine) Build(
|
||||
Func<(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine)> engineFactory)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(engineFactory);
|
||||
var pair = engineFactory();
|
||||
ArgumentNullException.ThrowIfNull(pair.Upstream, "engineFactory().Upstream");
|
||||
ArgumentNullException.ThrowIfNull(pair.Engine, "engineFactory().Engine");
|
||||
return pair;
|
||||
}
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="ScriptedAlarmHostActor"/> class.</summary>
|
||||
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
|
||||
/// <param name="mux">Optional dependency multiplexer the host registers dependency interest with.</param>
|
||||
@@ -183,14 +258,29 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
NodeId? localNode = null,
|
||||
Func<int>? driverMemberCountProvider = null,
|
||||
ITelemetryLocalHub? telemetryHub = null)
|
||||
: this(publishActor, mux, (upstream, engine), localNode, driverMemberCountProvider, telemetryHub, ownsEngine: false, commandTimeout: null)
|
||||
{
|
||||
}
|
||||
|
||||
private ScriptedAlarmHostActor(
|
||||
IActorRef publishActor,
|
||||
IActorRef? mux,
|
||||
(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine) pair,
|
||||
NodeId? localNode,
|
||||
Func<int>? driverMemberCountProvider,
|
||||
ITelemetryLocalHub? telemetryHub,
|
||||
bool ownsEngine,
|
||||
TimeSpan? commandTimeout)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(publishActor);
|
||||
ArgumentNullException.ThrowIfNull(upstream);
|
||||
ArgumentNullException.ThrowIfNull(engine);
|
||||
ArgumentNullException.ThrowIfNull(pair.Upstream);
|
||||
ArgumentNullException.ThrowIfNull(pair.Engine);
|
||||
_publishActor = publishActor;
|
||||
_mux = mux;
|
||||
_upstream = upstream;
|
||||
_engine = engine;
|
||||
_upstream = pair.Upstream;
|
||||
_engine = pair.Engine;
|
||||
_ownsEngine = ownsEngine;
|
||||
_commandTimeout = commandTimeout is { } t && t > TimeSpan.Zero ? t : DefaultCommandTimeout;
|
||||
_localNode = localNode;
|
||||
_driverMemberCountProvider = driverMemberCountProvider;
|
||||
_telemetryHub = telemetryHub;
|
||||
@@ -294,8 +384,15 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
// subscriptions exist, so any DependencyValueChanged the mux delivers will be observed by the
|
||||
// engine. RegisterInterest replaces a subscriber's prior interest set (see DependencyMuxActor),
|
||||
// so a re-Apply with a fresh union simply supersedes the old one — no explicit unregister needed.
|
||||
_mux?.Tell(new DependencyMuxActor.RegisterInterest(msg.DepRefs, Self));
|
||||
_log.Debug("ScriptedAlarmHost: loaded; registered mux interest for {Count} dep refs", msg.DepRefs.Count);
|
||||
// ReplayRetained: the mux replays the last value it saw for each ref, so a fresh engine (a
|
||||
// supervision restart builds one around an EMPTY upstream) converges immediately instead of
|
||||
// waiting for every input to change again.
|
||||
_mux?.Tell(new DependencyMuxActor.RegisterInterest(msg.DepRefs, Self, ReplayRetained: true));
|
||||
// Prune the upstream's retained snapshots to the loaded set (R02:L3): a path no alarm references
|
||||
// any more will never be pushed again, so its last value is dead weight for the process lifetime.
|
||||
var dropped = _upstream.RetainOnly(new HashSet<string>(msg.DepRefs, StringComparer.Ordinal));
|
||||
_log.Debug("ScriptedAlarmHost: loaded; registered mux interest for {Count} dep refs (pruned {Dropped} retained path(s))",
|
||||
msg.DepRefs.Count, dropped);
|
||||
|
||||
ReassertConditionNodes();
|
||||
}
|
||||
@@ -591,18 +688,25 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
return;
|
||||
}
|
||||
|
||||
// Bounded (R02:H3): every op ends in IAlarmStateStore.SaveAsync (LocalDb/SQLite), and this is a
|
||||
// ReceiveAsync — the mailbox is suspended until the op returns, so an unbounded await on a hung
|
||||
// file lock stalled dependency-value ingestion for the duration. Linked to the actor's lifetime
|
||||
// token so a stop still cancels promptly.
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token);
|
||||
cts.CancelAfter(_commandTimeout);
|
||||
var ct = cts.Token;
|
||||
try
|
||||
{
|
||||
switch (cmd.Operation)
|
||||
{
|
||||
case "Acknowledge":
|
||||
await _engine.AcknowledgeAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
|
||||
await _engine.AcknowledgeAsync(cmd.AlarmId, cmd.User, cmd.Comment, ct);
|
||||
break;
|
||||
case "Confirm":
|
||||
await _engine.ConfirmAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
|
||||
await _engine.ConfirmAsync(cmd.AlarmId, cmd.User, cmd.Comment, ct);
|
||||
break;
|
||||
case "OneShotShelve":
|
||||
await _engine.OneShotShelveAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
|
||||
await _engine.OneShotShelveAsync(cmd.AlarmId, cmd.User, ct);
|
||||
break;
|
||||
case "TimedShelve":
|
||||
// A timed shelve needs the absolute unshelve instant. T18 derives it from the OPC UA
|
||||
@@ -614,16 +718,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
cmd.AlarmId);
|
||||
return;
|
||||
}
|
||||
await _engine.TimedShelveAsync(cmd.AlarmId, cmd.User, unshelveAt, CancellationToken.None);
|
||||
await _engine.TimedShelveAsync(cmd.AlarmId, cmd.User, unshelveAt, ct);
|
||||
break;
|
||||
case "Unshelve":
|
||||
await _engine.UnshelveAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
|
||||
await _engine.UnshelveAsync(cmd.AlarmId, cmd.User, ct);
|
||||
break;
|
||||
case "Enable":
|
||||
await _engine.EnableAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
|
||||
await _engine.EnableAsync(cmd.AlarmId, cmd.User, ct);
|
||||
break;
|
||||
case "Disable":
|
||||
await _engine.DisableAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
|
||||
await _engine.DisableAsync(cmd.AlarmId, cmd.User, ct);
|
||||
break;
|
||||
case "AddComment":
|
||||
if (string.IsNullOrWhiteSpace(cmd.Comment))
|
||||
@@ -631,7 +735,7 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
_log.Warning("ScriptedAlarmHost: rejecting AddComment for {AlarmId} — empty comment text", cmd.AlarmId);
|
||||
return;
|
||||
}
|
||||
await _engine.AddCommentAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
|
||||
await _engine.AddCommentAsync(cmd.AlarmId, cmd.User, cmd.Comment, ct);
|
||||
break;
|
||||
default:
|
||||
_log.Warning("ScriptedAlarmHost: ignoring AlarmCommand with unknown operation {Op} for {AlarmId}",
|
||||
@@ -639,6 +743,11 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (!_cts.IsCancellationRequested)
|
||||
{
|
||||
_log.Warning("ScriptedAlarmHost: engine op {Op} for alarm {AlarmId} timed out after {Timeout} — the condition-state store is stalled",
|
||||
cmd.Operation, cmd.AlarmId, _commandTimeout);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// A failing engine op must not crash the actor — mirror OnLoadFailed's log-and-stay-inert style.
|
||||
@@ -672,16 +781,28 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
base.PreStart();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>Akka's default <c>PreRestart</c> stops the children and calls <see cref="PostStop"/>. The flag
|
||||
/// lets that <see cref="PostStop"/> keep a legacy (Props-captured) engine alive for the next incarnation
|
||||
/// instead of disposing the instance it is about to be re-created around.</remarks>
|
||||
protected override void PreRestart(Exception reason, object message)
|
||||
{
|
||||
_restarting = true;
|
||||
base.PreRestart(reason, message);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
// Unregister mux interest first so no further DependencyValueChanged arrives while we tear down.
|
||||
_mux?.Tell(new DependencyMuxActor.UnregisterInterest(Self));
|
||||
// Cancel any in-flight LoadAsync, detach the OnEvent handler (so a late engine emission can't
|
||||
// Tell a stopping actor), then dispose the engine (which drains its background work + clears).
|
||||
// Tell a stopping actor), then dispose the engine (which drains its background work + clears) —
|
||||
// unless this is a RESTART around a legacy Props-captured engine, which the next incarnation will
|
||||
// re-attach to (R02:H1).
|
||||
_cts.Cancel();
|
||||
_engine.OnEvent -= _onEngineEvent;
|
||||
_engine.Dispose();
|
||||
if (_ownsEngine || !_restarting) _engine.Dispose();
|
||||
_cts.Dispose();
|
||||
base.PostStop();
|
||||
}
|
||||
|
||||
@@ -62,7 +62,10 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor
|
||||
var refs = _consumer.DependencyRefs.ToList();
|
||||
// A RegisterInterest replaces the prior interest set for this subscriber (mux is idempotent on
|
||||
// re-register), so re-registering after an apply cleanly adopts an added/removed ref.
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(refs, Self));
|
||||
// ReplayRetained (R07:H6): a driver Restart is a NEW consumer behind a new adapter; the mux replays
|
||||
// the last value of every ref it retained so the calc tags compute from current inputs at once
|
||||
// instead of sitting BadWaitingForInitialData until each input happens to change.
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(refs, Self, ReplayRetained: true));
|
||||
_log.Debug("DependencyConsumerMuxAdapter: registered {Count} dependency ref(s)", refs.Count);
|
||||
}
|
||||
|
||||
|
||||
@@ -26,6 +26,20 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
/// <see cref="VirtualTagActor.DependencyValueChanged"/> receive as well, for direct senders and tests.
|
||||
/// The mux is deliberately not a DPS subscriber — virtual-tag evaluation is local to each node and
|
||||
/// would over-emit if it spanned the cluster.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Last-value retention + replay (R07:H6).</b> For every ref that has EVER been registered the
|
||||
/// mux retains the last <c>(value, status, timestamp)</c> it forwarded — through unregisters too,
|
||||
/// because the case that matters is a subscriber that went away and came back: a Calculation
|
||||
/// driver Restart is a new driver instance + a new mux adapter, and before this every calc tag sat
|
||||
/// <c>BadWaitingForInitialData</c> until each of its inputs happened to change again (a
|
||||
/// scripted-alarm host restart has the same shape). A <see cref="RegisterInterest"/> with
|
||||
/// <see cref="RegisterInterest.ReplayRetained"/> receives one batch of the retained values for its
|
||||
/// refs, in the same shape a driver burst would arrive in. Opt-in per registration: the
|
||||
/// continuous-historization recorder re-registers on every deploy and a replay there would
|
||||
/// record a stale sample per tag per deploy. Memory is bounded by the number of refs scripts have
|
||||
/// ever named on this node, not by the address space.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class DependencyMuxActor : ReceiveActor
|
||||
{
|
||||
@@ -33,14 +47,23 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
|
||||
/// <summary>Register a subscriber's interest in a set of tag refs. Idempotent on re-register —
|
||||
/// the second call replaces the prior interest set for that subscriber.</summary>
|
||||
public sealed record RegisterInterest(IReadOnlyList<string> TagRefs, IActorRef Subscriber);
|
||||
/// <param name="TagRefs">The refs the subscriber wants.</param>
|
||||
/// <param name="Subscriber">The subscriber.</param>
|
||||
/// <param name="ReplayRetained">When true, the mux immediately sends the subscriber one
|
||||
/// <see cref="VirtualTagActor.DependencyValueBatch"/> holding the last retained value of every ref
|
||||
/// in <paramref name="TagRefs"/> it has one for (nothing when it has none).</param>
|
||||
public sealed record RegisterInterest(IReadOnlyList<string> TagRefs, IActorRef Subscriber, bool ReplayRetained = false);
|
||||
|
||||
/// <summary>Unregister every interest held by <see cref="Subscriber"/>. Sent on PostStop by
|
||||
/// the subscriber, or by Terminated handling when the mux watches.</summary>
|
||||
public sealed record UnregisterInterest(IActorRef Subscriber);
|
||||
|
||||
// A ref stays in _byRef once registered — with an EMPTY set after its last subscriber leaves — so
|
||||
// the hot path's single lookup per item also decides retention. _byRef.Count == 0 keeps meaning
|
||||
// "nothing was ever registered", which the batch path early-outs on.
|
||||
private readonly Dictionary<string, HashSet<IActorRef>> _byRef = new(StringComparer.Ordinal);
|
||||
private readonly Dictionary<IActorRef, HashSet<string>> _bySubscriber = new();
|
||||
private readonly Dictionary<string, VirtualTagActor.DependencyValueChanged> _last = new(StringComparer.Ordinal);
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
|
||||
/// <summary>Creates props for the DependencyMuxActor.</summary>
|
||||
@@ -64,16 +87,13 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
{
|
||||
foreach (var r in priorRefs)
|
||||
{
|
||||
if (_byRef.TryGetValue(r, out var set))
|
||||
{
|
||||
set.Remove(msg.Subscriber);
|
||||
if (set.Count == 0) _byRef.Remove(r);
|
||||
}
|
||||
if (_byRef.TryGetValue(r, out var set)) set.Remove(msg.Subscriber);
|
||||
}
|
||||
}
|
||||
|
||||
var refs = new HashSet<string>(msg.TagRefs, StringComparer.Ordinal);
|
||||
_bySubscriber[msg.Subscriber] = refs;
|
||||
List<VirtualTagActor.DependencyValueChanged>? replay = null;
|
||||
foreach (var r in refs)
|
||||
{
|
||||
if (!_byRef.TryGetValue(r, out var set))
|
||||
@@ -82,9 +102,19 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
_byRef[r] = set;
|
||||
}
|
||||
set.Add(msg.Subscriber);
|
||||
if (msg.ReplayRetained && _last.TryGetValue(r, out var retained))
|
||||
(replay ??= []).Add(retained);
|
||||
}
|
||||
Context.Watch(msg.Subscriber);
|
||||
_log.Debug("DependencyMux: subscriber {Sub} registered for {Count} refs", msg.Subscriber, refs.Count);
|
||||
|
||||
// One batch, after the registration is complete, in exactly the shape a driver burst arrives in —
|
||||
// so the subscriber's hot path handles it with no special case.
|
||||
if (replay is not null)
|
||||
{
|
||||
msg.Subscriber.Tell(new VirtualTagActor.DependencyValueBatch(replay));
|
||||
_log.Debug("DependencyMux: replayed {Count} retained value(s) to {Sub}", replay.Count, msg.Subscriber);
|
||||
}
|
||||
}
|
||||
|
||||
private void RemoveSubscriber(IActorRef subscriber)
|
||||
@@ -92,11 +122,9 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
if (!_bySubscriber.TryGetValue(subscriber, out var refs)) return;
|
||||
foreach (var r in refs)
|
||||
{
|
||||
if (_byRef.TryGetValue(r, out var set))
|
||||
{
|
||||
set.Remove(subscriber);
|
||||
if (set.Count == 0) _byRef.Remove(r);
|
||||
}
|
||||
// The ref's entry stays (possibly empty) so its last value keeps being retained for a
|
||||
// subscriber that comes back — see the class remarks.
|
||||
if (_byRef.TryGetValue(r, out var set)) set.Remove(subscriber);
|
||||
}
|
||||
_bySubscriber.Remove(subscriber);
|
||||
Context.Unwatch(subscriber);
|
||||
@@ -105,13 +133,14 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
|
||||
private void OnAttributeValuePublished(DriverInstanceActor.AttributeValuePublished msg)
|
||||
{
|
||||
if (!_byRef.TryGetValue(msg.FullReference, out var subscribers) || subscribers.Count == 0)
|
||||
if (!_byRef.TryGetValue(msg.FullReference, out var subscribers))
|
||||
{
|
||||
// No virtual tag cares about this ref — drop. Common in normal operation; the address
|
||||
// No virtual tag ever cared about this ref — drop. Common in normal operation; the address
|
||||
// space carries thousands of tags and only a fraction feed virtual-tag expressions.
|
||||
return;
|
||||
}
|
||||
var dep = new VirtualTagActor.DependencyValueChanged(msg.FullReference, msg.Value, msg.TimestampUtc, msg.StatusCode);
|
||||
_last[msg.FullReference] = dep;
|
||||
foreach (var sub in subscribers)
|
||||
{
|
||||
// A single-item batch, so the mux emits exactly ONE message shape and every subscriber needs
|
||||
@@ -122,15 +151,6 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Filters one batched driver production burst per item and delivers each subscriber's matched
|
||||
/// slice as ONE <see cref="VirtualTagActor.DependencyValueBatch"/>. Items nobody registered for are
|
||||
/// dropped (the common case: only a fraction of an address space feeds virtual-tag expressions), so a
|
||||
/// burst of N values costs one inbound message plus one outbound message per genuinely interested
|
||||
/// SUBSCRIBER — not per (item, subscriber) pair as it did when this fanned out per value. That is what
|
||||
/// lets <see cref="VirtualTagActor"/> evaluate its script once per burst instead of once per input.
|
||||
/// <para>The registered set is NOT just virtual-tag inputs: <c>ContinuousHistorizationRecorder</c>
|
||||
/// registers interest in every historized ref, so on a historizing node <see cref="_byRef"/> can be
|
||||
/// large and a sizeable fraction of each burst resolves.</para></summary>
|
||||
private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
// Nobody has registered anything — skip the whole burst rather than probing an empty map per item.
|
||||
@@ -146,8 +166,10 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
for (var i = 0; i < count; i++)
|
||||
{
|
||||
var item = msg.Items[i];
|
||||
if (!_byRef.TryGetValue(item.FullReference, out var subscribers) || subscribers.Count == 0) continue;
|
||||
if (!_byRef.TryGetValue(item.FullReference, out var subscribers)) continue;
|
||||
var dep = new VirtualTagActor.DependencyValueChanged(item.FullReference, item.Value, item.TimestampUtc, item.StatusCode);
|
||||
_last[item.FullReference] = dep;
|
||||
if (subscribers.Count == 0) continue; // retained for a returning subscriber; nobody to forward to now
|
||||
perSubscriber ??= new Dictionary<IActorRef, List<VirtualTagActor.DependencyValueChanged>>();
|
||||
foreach (var sub in subscribers)
|
||||
{
|
||||
|
||||
@@ -7,6 +7,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
@@ -76,6 +77,10 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
private readonly IReadOnlyList<string> _dependencyRefs;
|
||||
private readonly IActorRef? _mux;
|
||||
private readonly ITelemetryLocalHub? _telemetryHub;
|
||||
// R02:P6 — a script failing on every trigger of a 1 Hz input published one cluster-wide DPS
|
||||
// ScriptLogEntry per second per tag; 200 such tags shed everyone's entries out of the bounded
|
||||
// telemetry channels. The per-failure entry is now rate-limited per tag (20/min + one summary).
|
||||
private readonly ScriptLogRateLimiter _failureLogLimiter;
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
// Dependency SNAPSHOTS, not bare values (archreview 02/S14): the status code is what distinguishes a
|
||||
// comms-lost input from a genuine Good null, and it is what the script reads through
|
||||
@@ -104,6 +109,8 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
/// <param name="mux">Optional reference to a dependency multiplexer actor.</param>
|
||||
/// <param name="telemetryHub">Optional node-local live-telemetry hub each script-log entry is also
|
||||
/// emitted into (Phase 5); null (tests) simply skips the emit.</param>
|
||||
/// <param name="failureLogLimiter">Rate limiter for this tag's per-failure script-log entries; null ⇒
|
||||
/// the default 20 per minute + one summary.</param>
|
||||
/// <returns>A configured <see cref="Akka.Actor.Props"/> for creating the actor.</returns>
|
||||
public static Props Props(
|
||||
string virtualTagId,
|
||||
@@ -113,7 +120,8 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
Func<DPSPublisher>? publisherFactory = null,
|
||||
IReadOnlyList<string>? dependencyRefs = null,
|
||||
IActorRef? mux = null,
|
||||
ITelemetryLocalHub? telemetryHub = null) =>
|
||||
ITelemetryLocalHub? telemetryHub = null,
|
||||
ScriptLogRateLimiter? failureLogLimiter = null) =>
|
||||
Akka.Actor.Props.Create(() => new VirtualTagActor(
|
||||
virtualTagId, expression,
|
||||
evaluator ?? NullVirtualTagEvaluator.Instance,
|
||||
@@ -121,7 +129,8 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
publisherFactory,
|
||||
dependencyRefs ?? Array.Empty<string>(),
|
||||
mux,
|
||||
telemetryHub));
|
||||
telemetryHub,
|
||||
failureLogLimiter));
|
||||
|
||||
/// <summary>Initializes a virtual tag actor with the given configuration and dependencies.</summary>
|
||||
/// <param name="virtualTagId">Unique identifier for the virtual tag.</param>
|
||||
@@ -133,6 +142,8 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
/// <param name="mux">Optional reference to a dependency multiplexer actor.</param>
|
||||
/// <param name="telemetryHub">Optional node-local live-telemetry hub each script-log entry is also
|
||||
/// emitted into (Phase 5); null (tests) simply skips the emit.</param>
|
||||
/// <param name="failureLogLimiter">Rate limiter for this tag's per-failure script-log entries; null ⇒
|
||||
/// the default 20 per minute + one summary.</param>
|
||||
public VirtualTagActor(
|
||||
string virtualTagId,
|
||||
string expression,
|
||||
@@ -141,7 +152,8 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
Func<DPSPublisher>? publisherFactory,
|
||||
IReadOnlyList<string> dependencyRefs,
|
||||
IActorRef? mux,
|
||||
ITelemetryLocalHub? telemetryHub = null)
|
||||
ITelemetryLocalHub? telemetryHub = null,
|
||||
ScriptLogRateLimiter? failureLogLimiter = null)
|
||||
{
|
||||
_virtualTagId = virtualTagId;
|
||||
_scriptId = scriptId;
|
||||
@@ -151,6 +163,7 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
_dependencyRefs = dependencyRefs;
|
||||
_mux = mux;
|
||||
_telemetryHub = telemetryHub;
|
||||
_failureLogLimiter = failureLogLimiter ?? new ScriptLogRateLimiter();
|
||||
|
||||
Receive<DependencyValueChanged>(OnDependencyChanged);
|
||||
Receive<DependencyValueBatch>(OnDependencyBatch);
|
||||
@@ -162,7 +175,9 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
{
|
||||
if (_mux is not null && _dependencyRefs.Count > 0)
|
||||
{
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(_dependencyRefs, Self));
|
||||
// ReplayRetained: a respawned child (plan edit, supervision restart) evaluates from the mux's
|
||||
// last-known inputs at once instead of waiting for each to change again.
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(_dependencyRefs, Self, ReplayRetained: true));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -353,7 +368,18 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
private void OnEvaluationFailed(string reason, string level, DateTime timestampUtc)
|
||||
{
|
||||
OtOpcUaTelemetry.VirtualTagEval.Add(1, new KeyValuePair<string, object?>("outcome", "fail"));
|
||||
PublishLog(level, reason);
|
||||
|
||||
// Rate-limited per tag (R02:P6): the first 20 failures a minute publish as before; the rest are
|
||||
// counted and surfaced as ONE summary entry when the window rolls. The metric above and the Bad
|
||||
// publish below are not limited — only the cluster-wide log fan-out is.
|
||||
var admission = _failureLogLimiter.Admit(_virtualTagId);
|
||||
if (admission.SuppressedInPreviousWindow > 0)
|
||||
{
|
||||
PublishLog("Warning",
|
||||
$"{admission.SuppressedInPreviousWindow} further evaluation failure(s) were suppressed from the script log " +
|
||||
$"in the previous {_failureLogLimiter.Window.TotalSeconds:F0} s (limit {ScriptLogRateLimiter.DefaultLimitPerWindow}/window); last: {reason}");
|
||||
}
|
||||
if (admission.Admitted) PublishLog(level, reason);
|
||||
|
||||
if ((_hasPublished && _lastPublishedStatus == OpcUaStatus.BadInternalError) || !AllDependenciesArrived())
|
||||
{
|
||||
|
||||
@@ -82,7 +82,9 @@ public sealed class DraftValidatorCalculationTests
|
||||
DriverInstances = [CalcDriver()],
|
||||
Devices = [Device("dev-1", "di-calc", "engine")],
|
||||
Tags = [CalcTagRow("t-1", "dev-1", "a", Cfg("s1"))],
|
||||
Scripts = [ScriptRow("s1", "return 42.0;")],
|
||||
// A read, so the change trigger is effective (a constant change-triggered script with no timer
|
||||
// is now itself a deploy error — CalculationNoEffectiveTrigger).
|
||||
Scripts = [ScriptRow("s1", "return (double)ctx.GetTag(\"mb/plc/reg\").Value + 42.0;")],
|
||||
};
|
||||
|
||||
DraftValidator.Validate(draft).ShouldNotContain(e => e.Code.StartsWith("Calculation"));
|
||||
@@ -132,6 +134,58 @@ public sealed class DraftValidatorCalculationTests
|
||||
msg.ShouldContain("calc/engine/b");
|
||||
}
|
||||
|
||||
// ── effective-trigger gate (R07:U5) ─────────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public void Change_triggered_tag_with_no_literal_read_and_no_timer_is_rejected()
|
||||
{
|
||||
var draft = new DraftSnapshot
|
||||
{
|
||||
GenerationId = 1, ClusterId = "c",
|
||||
DriverInstances = [CalcDriver()],
|
||||
Devices = [Device("dev-1", "di-calc", "engine")],
|
||||
Tags = [CalcTagRow("t-1", "dev-1", "a", Cfg("s1"))],
|
||||
Scripts = [ScriptRow("s1", "return ctx.Now.Hour;")],
|
||||
};
|
||||
|
||||
DraftValidator.Validate(draft).ShouldContain(e =>
|
||||
e.Code == "CalculationNoEffectiveTrigger" && e.Context == "t-1" && e.Message.Contains("no literal ctx.GetTag"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Tag_with_change_trigger_off_and_no_timer_is_rejected()
|
||||
{
|
||||
var draft = new DraftSnapshot
|
||||
{
|
||||
GenerationId = 1, ClusterId = "c",
|
||||
DriverInstances = [CalcDriver()],
|
||||
Devices = [Device("dev-1", "di-calc", "engine")],
|
||||
Tags = [CalcTagRow("t-1", "dev-1", "a", "{\"scriptId\":\"s1\",\"changeTriggered\":false}")],
|
||||
Scripts = [ScriptRow("s1", "return (double)ctx.GetTag(\"mb/plc/reg\").Value;")],
|
||||
};
|
||||
|
||||
DraftValidator.Validate(draft).ShouldContain(e => e.Code == "CalculationNoEffectiveTrigger" && e.Context == "t-1");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("{\"scriptId\":\"s1\",\"changeTriggered\":true,\"timerIntervalMs\":1000}", "return ctx.Now.Hour;")] // timer rescues a read-less script
|
||||
[InlineData("{\"scriptId\":\"s1\",\"changeTriggered\":false,\"timerIntervalMs\":500}", "return 1.0;")] // timer-only tag
|
||||
[InlineData("{\"scriptId\":\"s1\",\"changeTriggered\":true}", "return (double)ctx.GetTag(\"mb/plc/reg\").Value;")] // change trigger with a read
|
||||
[InlineData("{\"scriptId\":\"s1\"}", "return (double)ctx.GetTag(\"mb/plc/reg\").Value;")] // changeTriggered defaults to true
|
||||
public void A_tag_with_an_effective_trigger_is_accepted(string cfg, string source)
|
||||
{
|
||||
var draft = new DraftSnapshot
|
||||
{
|
||||
GenerationId = 1, ClusterId = "c",
|
||||
DriverInstances = [CalcDriver()],
|
||||
Devices = [Device("dev-1", "di-calc", "engine")],
|
||||
Tags = [CalcTagRow("t-1", "dev-1", "a", cfg)],
|
||||
Scripts = [ScriptRow("s1", source)],
|
||||
};
|
||||
|
||||
DraftValidator.Validate(draft).ShouldNotContain(e => e.Code == "CalculationNoEffectiveTrigger");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Cross_driver_ref_is_terminal_not_an_edge()
|
||||
{
|
||||
|
||||
+166
@@ -0,0 +1,166 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Serilog;
|
||||
using Serilog.Events;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Review 07 Calculation remediation: a compile failure is negative-cached by source hash and its
|
||||
/// Warning/script-log fires on the Good→Bad transition only (P4); inputs are recorded before
|
||||
/// <c>InitializeAsync</c> so a Restart computes from current inputs (H6); array results dedup
|
||||
/// structurally (H8); <c>ShutdownAsync</c> drops the compiled scripts (L2).
|
||||
/// </summary>
|
||||
public sealed class CalculationDriverRemediationTests
|
||||
{
|
||||
private const uint StatusGood = 0u;
|
||||
private const uint StatusBadInternalError = 0x80020000u;
|
||||
|
||||
private static RawTagEntry CalcTag(string rawPath, string source, bool changeTriggered = true, int? timerMs = null)
|
||||
{
|
||||
var timer = timerMs is { } ms ? $",\"timerIntervalMs\":{ms}" : "";
|
||||
var json = $"{{\"scriptId\":\"s-{rawPath}\",\"changeTriggered\":{(changeTriggered ? "true" : "false")}{timer}," +
|
||||
$"\"scriptSource\":{System.Text.Json.JsonSerializer.Serialize(source)}}}";
|
||||
return new RawTagEntry(rawPath, json, WriteIdempotent: false);
|
||||
}
|
||||
|
||||
private static (CalculationDriver Driver, ConcurrentQueue<DataChangeEventArgs> Published, List<ScriptLogEntry> Logs, CapturingLogger HostLog)
|
||||
Build(params RawTagEntry[] tags)
|
||||
{
|
||||
var logs = new List<ScriptLogEntry>();
|
||||
var sink = new ScriptLogTopicSink(new CapturingPublisher(logs), LogEventLevel.Information);
|
||||
var scriptRoot = new ScriptRootLogger(new LoggerConfiguration().WriteTo.Sink(sink).CreateLogger());
|
||||
var hostLog = new CapturingLogger();
|
||||
var driver = new CalculationDriver(new CalculationDriverOptions { RawTags = tags }, "calc-drv", scriptRoot, hostLog);
|
||||
var published = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
driver.OnDataChange += (_, e) => published.Enqueue(e);
|
||||
return (driver, published, logs, hostLog);
|
||||
}
|
||||
|
||||
private static async Task Settle(CalculationDriver driver)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddSeconds(30);
|
||||
while (!driver.IsEvaluationIdle && DateTime.UtcNow < deadline) await Task.Delay(10);
|
||||
driver.IsEvaluationIdle.ShouldBeTrue("the calc driver's evaluation pump should have drained");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_compile_failure_is_remembered_and_logged_once_across_repeated_triggers()
|
||||
{
|
||||
var (driver, published, logs, hostLog) = Build(
|
||||
CalcTag("calc/e/broken", "var x = (double)ctx.GetTag(\"src/a\").Value; this is not valid C#;"));
|
||||
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
|
||||
for (var i = 0; i < 5; i++)
|
||||
{
|
||||
driver.OnDependencyValue("src/a", i, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
}
|
||||
|
||||
published.Count.ShouldBe(1, "Bad is published once per transition");
|
||||
published.TryDequeue(out var bad).ShouldBeTrue();
|
||||
bad!.Snapshot.StatusCode.ShouldBe(StatusBadInternalError);
|
||||
logs.Count.ShouldBe(1, "the script-log entry fires on the Good→Bad transition only");
|
||||
hostLog.Entries.Count(e => e.Level == LogLevel.Warning && e.Message.Contains("compile failed")).ShouldBe(1,
|
||||
"Roslyn is consulted once per source; later triggers hit the negative cache");
|
||||
driver.CompileFailureCountForTest.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_changed_failure_reason_is_logged_even_while_already_Bad()
|
||||
{
|
||||
// Throws with a different message depending on the input, so the failure REASON changes while the
|
||||
// published status stays Bad — the operator must see the new reason.
|
||||
var (driver, published, logs, _) = Build(
|
||||
CalcTag("calc/e/thrower", "throw new System.InvalidOperationException(\"boom \" + (int)ctx.GetTag(\"src/a\").Value);"));
|
||||
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
|
||||
driver.OnDependencyValue("src/a", 1, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
driver.OnDependencyValue("src/a", 1, StatusGood, DateTime.UtcNow); // same reason → no new log
|
||||
await Settle(driver);
|
||||
driver.OnDependencyValue("src/a", 2, StatusGood, DateTime.UtcNow); // new reason → logged
|
||||
await Settle(driver);
|
||||
|
||||
published.Count.ShouldBe(1);
|
||||
logs.Count.ShouldBe(2);
|
||||
logs[0].Message.ShouldContain("boom 1");
|
||||
logs[1].Message.ShouldContain("boom 2");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Inputs_delivered_before_InitializeAsync_are_kept_so_the_first_trigger_computes()
|
||||
{
|
||||
// H6: a Restart is a new driver whose adapter registers interest at construction; the mux's
|
||||
// retained-value replay therefore lands BEFORE InitializeAsync. It used to be dropped.
|
||||
var (driver, published, _, _) = Build(
|
||||
CalcTag("calc/e/sum", "return (double)ctx.GetTag(\"src/a\").Value + (double)ctx.GetTag(\"src/b\").Value;"));
|
||||
|
||||
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow); // replayed, pre-init
|
||||
driver.OnDependencyValue("src/b", 20.0, StatusGood, DateTime.UtcNow);
|
||||
published.ShouldBeEmpty("evaluation is gated on running");
|
||||
|
||||
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
driver.OnDependencyValue("src/a", 12.0, StatusGood, DateTime.UtcNow); // only ONE input moves after init
|
||||
await Settle(driver);
|
||||
|
||||
published.Count.ShouldBe(1, "src/b's pre-init value satisfied the change-gate");
|
||||
published.TryDequeue(out var e).ShouldBeTrue();
|
||||
e!.Snapshot.Value.ShouldBe(32.0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_equal_array_result_is_deduped_structurally()
|
||||
{
|
||||
var (driver, published, _, _) = Build(
|
||||
CalcTag("calc/e/arr", "var v = (double)ctx.GetTag(\"src/a\").Value; return new double[] { v, v * 2 };"));
|
||||
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
|
||||
driver.OnDependencyValue("src/a", 1.0, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
driver.OnDependencyValue("src/a", 1.0, StatusGood, DateTime.UtcNow); // same array contents
|
||||
await Settle(driver);
|
||||
published.Count.ShouldBe(1, "a structurally equal array is not a change");
|
||||
|
||||
driver.OnDependencyValue("src/a", 2.0, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
published.Count.ShouldBe(2, "a different array still publishes");
|
||||
published.Last().Snapshot.Value.ShouldBe(new double[] { 2.0, 4.0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ShutdownAsync_drops_the_compiled_scripts()
|
||||
{
|
||||
var (driver, published, _, _) = Build(CalcTag("calc/e/x", "return (double)ctx.GetTag(\"src/a\").Value * 2.0;"));
|
||||
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
driver.OnDependencyValue("src/a", 1.0, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
published.Count.ShouldBe(1);
|
||||
driver.CompiledScriptCountForTest.ShouldBe(1);
|
||||
|
||||
await driver.ShutdownAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
driver.CompiledScriptCountForTest.ShouldBe(0, "the collectible ALCs must not survive the generation");
|
||||
}
|
||||
|
||||
private sealed class CapturingPublisher(List<ScriptLogEntry> entries) : IScriptLogPublisher
|
||||
{
|
||||
public void Publish(ScriptLogEntry entry) { lock (entries) entries.Add(entry); }
|
||||
}
|
||||
|
||||
internal sealed class CapturingLogger : ILogger<CalculationDriver>
|
||||
{
|
||||
public ConcurrentBag<(LogLevel Level, string Message)> Entries { get; } = new();
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
=> Entries.Add((logLevel, formatter(state, exception)));
|
||||
}
|
||||
}
|
||||
@@ -229,11 +229,13 @@ public sealed class CalculationDriverTests
|
||||
logs[0].ScriptId.ShouldBe("s-calc/e/div");
|
||||
logs[0].VirtualTagId.ShouldBe("calc/e/div");
|
||||
|
||||
// a = 0 again → still failing, but NOT a fresh transition → no second Bad publish (script-log still emits).
|
||||
// a = 0 again → still failing for the SAME reason, NOT a fresh transition → no second Bad publish,
|
||||
// and (R07:P4) no second script-log entry either: the log fires on the transition or on a change of
|
||||
// reason, not on every trigger of a tag that is already Bad.
|
||||
driver.OnDependencyValue("src/a", 0, StatusGood, DateTime.UtcNow);
|
||||
await Settle(driver);
|
||||
published.ShouldBeEmpty();
|
||||
logs.Count.ShouldBe(2);
|
||||
logs.Count.ShouldBe(1);
|
||||
|
||||
// a = 5 → 20 → Good recovery is force-published.
|
||||
driver.OnDependencyValue("src/a", 5, StatusGood, DateTime.UtcNow);
|
||||
|
||||
+22
@@ -15,6 +15,28 @@ public sealed class DependencyMuxTagUpstreamSourceTests
|
||||
private static DataValueSnapshot Good(object? value, DateTime ts)
|
||||
=> new(value, 0u, ts, ts);
|
||||
|
||||
/// <summary>R02:L3 — the host prunes the retained snapshots to the loaded dependency-ref union after each
|
||||
/// (re)load, so a renamed/removed raw path stops being retained; observers are untouched.</summary>
|
||||
[Fact]
|
||||
public void RetainOnly_drops_snapshots_outside_the_set_and_keeps_observers()
|
||||
{
|
||||
var src = new DependencyMuxTagUpstreamSource();
|
||||
var ts = DateTime.UtcNow;
|
||||
src.Push("keep", Good(1, ts));
|
||||
src.Push("old/renamed", Good(2, ts));
|
||||
var seen = new List<string>();
|
||||
using var sub = src.SubscribeTag("old/renamed", (p, _) => seen.Add(p));
|
||||
|
||||
src.RetainOnly(new HashSet<string>(StringComparer.Ordinal) { "keep" }).ShouldBe(1);
|
||||
|
||||
src.RetainedPathCount.ShouldBe(1);
|
||||
src.ReadTag("keep").Value.ShouldBe(1);
|
||||
src.ReadTag("old/renamed").StatusCode.ShouldBe(OpcUaStatus.BadWaitingForInitialData);
|
||||
src.Push("old/renamed", Good(3, ts));
|
||||
// Pruning retained values does not unsubscribe anyone.
|
||||
seen.ShouldBe(new[] { "old/renamed" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Push_then_ReadTag_returns_the_pushed_snapshot()
|
||||
{
|
||||
|
||||
+145
@@ -0,0 +1,145 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit;
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.ScriptedAlarms;
|
||||
|
||||
/// <summary>
|
||||
/// R02:H1 — a supervision restart of <see cref="ScriptedAlarmHostActor"/> used to dispose the
|
||||
/// Props-captured engine and re-create the actor around the disposed instance, so every later
|
||||
/// <c>ApplyScriptedAlarms</c> failed and the node's scripted alarms silently died. With the factory
|
||||
/// Props each incarnation builds its own engine; with the legacy Props a restart only detaches. Both
|
||||
/// shapes must load cleanly after a throw inside a <see cref="VirtualTagActor.DependencyValueBatch"/>
|
||||
/// handler. Also R02:H3 — an inbound alarm command's engine op is bounded.
|
||||
/// </summary>
|
||||
public sealed class ScriptedAlarmHostActorRestartTests : RuntimeActorTestBase
|
||||
{
|
||||
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
private static EquipmentScriptedAlarmPlan Plan(string id = "alm-1", string depRef = "M.T") =>
|
||||
new(ScriptedAlarmId: id, EquipmentId: "Plant/Line1/M", Name: "HighTemp", AlarmType: "AlarmCondition",
|
||||
Severity: 800, MessageTemplate: "condition", PredicateScriptId: $"{id}-script",
|
||||
PredicateSource: $"return (int)ctx.GetTag(\"{depRef}\").Value > 90;",
|
||||
DependencyRefs: new[] { depRef }, HistorizeToAveva: true, Retain: true, Enabled: true);
|
||||
|
||||
private static ScriptedAlarmEngine BuildEngine(DependencyMuxTagUpstreamSource upstream, IAlarmStateStore? store = null)
|
||||
{
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
return new ScriptedAlarmEngine(upstream, store ?? new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
|
||||
}
|
||||
|
||||
private static (DependencyMuxTagUpstreamSource, ScriptedAlarmEngine) Pair()
|
||||
{
|
||||
var up = new DependencyMuxTagUpstreamSource();
|
||||
return (up, BuildEngine(up));
|
||||
}
|
||||
|
||||
/// <summary>Apply, crash the handler (a null TagId makes <c>Push</c> throw → restart), apply again
|
||||
/// and prove the second load registers interest and the alarm still activates.</summary>
|
||||
private void RestartThenReload(IActorRef host, TestProbe publish, TestProbe mux)
|
||||
{
|
||||
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan() }));
|
||||
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
|
||||
|
||||
// The crash: a batch whose item has a null TagId makes the upstream Push throw out of the handler
|
||||
// (the documented, deliberately unguarded path) → supervision restarts the actor.
|
||||
host.Tell(new VirtualTagActor.DependencyValueBatch(new[]
|
||||
{
|
||||
new VirtualTagActor.DependencyValueChanged(null!, 1, DateTime.UtcNow),
|
||||
}));
|
||||
mux.ExpectMsg<DependencyMuxActor.UnregisterInterest>(Timeout); // PostStop of the crashed incarnation
|
||||
|
||||
// A later apply must load cleanly on the restarted incarnation…
|
||||
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan() }));
|
||||
var reg = mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
|
||||
reg.TagRefs.ShouldContain("M.T");
|
||||
|
||||
// …and the alarm must still work end to end.
|
||||
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
|
||||
var state = publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout);
|
||||
state.AlarmNodeId.ShouldBe("alm-1");
|
||||
state.State.Active.ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Factory_Props_survives_a_handler_throw_and_loads_cleanly_afterwards()
|
||||
{
|
||||
var publish = CreateTestProbe();
|
||||
var mux = CreateTestProbe();
|
||||
var built = 0;
|
||||
var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(publish.Ref, mux.Ref, () => { built++; return Pair(); }));
|
||||
|
||||
RestartThenReload(host, publish, mux);
|
||||
|
||||
built.ShouldBe(2, "the restarted incarnation built its own engine");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Legacy_instance_Props_survives_a_handler_throw_because_a_restart_only_detaches()
|
||||
{
|
||||
var publish = CreateTestProbe();
|
||||
var mux = CreateTestProbe();
|
||||
var (upstream, engine) = Pair();
|
||||
var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(publish.Ref, mux.Ref, upstream, engine));
|
||||
|
||||
RestartThenReload(host, publish, mux);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_alarm_command_whose_store_hangs_is_bounded_and_the_mailbox_moves_on()
|
||||
{
|
||||
var publish = CreateTestProbe();
|
||||
var mux = CreateTestProbe();
|
||||
var store = new HangingSaveStore();
|
||||
var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(publish.Ref, mux.Ref,
|
||||
() => { var up = new DependencyMuxTagUpstreamSource(); return (up, BuildEngine(up, store)); },
|
||||
commandTimeout: TimeSpan.FromMilliseconds(300)));
|
||||
|
||||
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan() }));
|
||||
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
|
||||
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
|
||||
publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout).State.Active.ShouldBeTrue();
|
||||
|
||||
store.HangNext = true;
|
||||
host.Tell(new AlarmCommand("alm-1", "Acknowledge", "alice", "ack", null));
|
||||
|
||||
// The hung save would have suspended the mailbox indefinitely; the bound releases it, so the
|
||||
// next dependency value is processed and the alarm clears.
|
||||
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 10, DateTime.UtcNow));
|
||||
publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout).State.Active.ShouldBeFalse();
|
||||
store.ObservedCancellation.ShouldBeTrue();
|
||||
}
|
||||
|
||||
/// <summary>Store whose next SaveAsync blocks until its token is cancelled.</summary>
|
||||
private sealed class HangingSaveStore : IAlarmStateStore
|
||||
{
|
||||
private readonly InMemoryAlarmStateStore _inner = new();
|
||||
public volatile bool HangNext;
|
||||
public volatile bool ObservedCancellation;
|
||||
|
||||
public Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct) => _inner.LoadAsync(alarmId, ct);
|
||||
public Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct) => _inner.LoadAllAsync(ct);
|
||||
public Task RemoveAsync(string alarmId, CancellationToken ct) => _inner.RemoveAsync(alarmId, ct);
|
||||
|
||||
public async Task SaveAsync(AlarmConditionState state, CancellationToken ct)
|
||||
{
|
||||
if (HangNext)
|
||||
{
|
||||
HangNext = false;
|
||||
try { await Task.Delay(System.Threading.Timeout.InfiniteTimeSpan, ct); }
|
||||
catch (OperationCanceledException) { ObservedCancellation = true; throw; }
|
||||
}
|
||||
await _inner.SaveAsync(state, ct);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Calculation;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// R07:H6 — the mux retains the last value per ever-registered ref and replays it to a registration
|
||||
/// that asks (<see cref="DependencyMuxActor.RegisterInterest.ReplayRetained"/>). A Calculation driver
|
||||
/// Restart (new driver + new adapter) therefore republishes from current inputs instead of sitting
|
||||
/// <c>BadWaitingForInitialData</c> until every input happens to change.
|
||||
/// </summary>
|
||||
public sealed class DependencyMuxReplayTests : RuntimeActorTestBase
|
||||
{
|
||||
private static DriverInstanceActor.AttributeValueBatchPublished Burst(params (string Ref, object? Value)[] items) =>
|
||||
new("driver-1", items.Select(i => new DriverInstanceActor.AttributeValueItem(i.Ref, i.Value, OpcUaStatus.Good, DateTime.UtcNow)).ToArray());
|
||||
|
||||
[Fact]
|
||||
public void RegisterInterest_with_replay_receives_the_retained_values_for_its_refs_only()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var first = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "b" }, first.Ref));
|
||||
mux.Tell(Burst(("a", 1), ("b", 2), ("c", 3)));
|
||||
first.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget).Items.Count.ShouldBe(2);
|
||||
|
||||
// The first subscriber goes away; retention must survive the unregister (that is the Restart shape).
|
||||
mux.Tell(new DependencyMuxActor.UnregisterInterest(first.Ref));
|
||||
|
||||
var second = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "c" }, second.Ref, ReplayRetained: true));
|
||||
|
||||
var replay = second.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget);
|
||||
replay.Items.Select(i => i.TagId).ShouldBe(new[] { "a" }, "'c' was never registered by anyone, so it is not retained; 'b' is not in this subscriber's refs");
|
||||
replay.Items[0].Value.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RegisterInterest_without_replay_receives_nothing_and_a_fresh_ref_replays_nothing()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var seed = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a" }, seed.Ref));
|
||||
mux.Tell(Burst(("a", 1)));
|
||||
seed.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget);
|
||||
|
||||
var opted = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a" }, opted.Ref)); // default: no replay
|
||||
opted.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
|
||||
var fresh = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "never-published" }, fresh.Ref, ReplayRetained: true));
|
||||
fresh.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
}
|
||||
|
||||
/// <summary>The user-facing scenario: a calc tag computes, the driver is restarted (new instance + new
|
||||
/// adapter) with NO input changing afterwards, and the new instance republishes the current value.</summary>
|
||||
[Fact]
|
||||
public async Task A_Calculation_driver_restart_republishes_from_the_current_inputs()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var tag = new RawTagEntry("calc/e/sum",
|
||||
"{\"scriptId\":\"s\",\"changeTriggered\":true,\"scriptSource\":\"return (double)ctx.GetTag(\\\"src/a\\\").Value + (double)ctx.GetTag(\\\"src/b\\\").Value;\"}",
|
||||
WriteIdempotent: false);
|
||||
|
||||
// Generation 1: wire the driver through the real adapter and feed a burst.
|
||||
var gen1 = new CalculationDriver(new CalculationDriverOptions { RawTags = [tag] }, "calc-drv");
|
||||
var published1 = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
gen1.OnDataChange += (_, e) => published1.Enqueue(e);
|
||||
await gen1.InitializeAsync("{}", CancellationToken.None);
|
||||
var adapter1 = Sys.ActorOf(DependencyConsumerMuxAdapter.Props(gen1, mux));
|
||||
AwaitAssert(() =>
|
||||
{
|
||||
mux.Tell(Burst(("src/a", 10.0), ("src/b", 20.0)));
|
||||
published1.Count.ShouldBeGreaterThan(0);
|
||||
}, PresenceBudget);
|
||||
published1.Last().Snapshot.Value.ShouldBe(30.0);
|
||||
|
||||
// Restart: the old instance goes away; a NEW driver + adapter come up. Nothing changes upstream.
|
||||
Sys.Stop(adapter1);
|
||||
await gen1.ShutdownAsync(CancellationToken.None);
|
||||
var gen2 = new CalculationDriver(new CalculationDriverOptions { RawTags = [tag] }, "calc-drv");
|
||||
var published2 = new ConcurrentQueue<DataChangeEventArgs>();
|
||||
gen2.OnDataChange += (_, e) => published2.Enqueue(e);
|
||||
Sys.ActorOf(DependencyConsumerMuxAdapter.Props(gen2, mux)); // registers with ReplayRetained
|
||||
await gen2.InitializeAsync("{}", CancellationToken.None);
|
||||
|
||||
// The replayed inputs satisfied the change-gate; the first trigger after init is the replay itself
|
||||
// (if it lands post-init) or the next change (here: none) — so publish happens from the replay.
|
||||
AwaitAssert(() => published2.Count.ShouldBeGreaterThan(0), PresenceBudget);
|
||||
published2.Last().Snapshot.Value.ShouldBe(30.0);
|
||||
published2.Last().Snapshot.StatusCode.ShouldBe(OpcUaStatus.Good);
|
||||
}
|
||||
}
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
|
||||
|
||||
/// <summary>R02:P6 — the per-failure cluster-wide <see cref="ScriptLogEntry"/> is rate-limited per tag
|
||||
/// (first N per window, then one summary); the Bad publish and the metric are unaffected.</summary>
|
||||
public sealed class VirtualTagActorFailureLogRateLimitTests : RuntimeActorTestBase
|
||||
{
|
||||
[Fact]
|
||||
public void Repeated_failures_publish_at_most_the_limit_then_one_summary_when_the_window_rolls()
|
||||
{
|
||||
var now = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc);
|
||||
var limiter = new ScriptLogRateLimiter(limitPerWindow: 3, window: TimeSpan.FromMinutes(1), clock: () => now);
|
||||
var capture = new ConcurrentQueue<ScriptLogEntry>();
|
||||
var parent = CreateTestProbe();
|
||||
var actor = parent.ChildActorOf(VirtualTagActor.Props(
|
||||
"vt-chatty", "broken",
|
||||
evaluator: new FailingEvaluator("boom"),
|
||||
publisherFactory: () => new DPSPublisher((_, p) => capture.Enqueue((ScriptLogEntry)p)),
|
||||
failureLogLimiter: limiter));
|
||||
|
||||
for (var i = 0; i < 10; i++)
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", i, DateTime.UtcNow));
|
||||
|
||||
// Positive control: the FIRST failure still degrades the node (once per transition, as before).
|
||||
parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget).StatusCode.ShouldBe(Commons.OpcUa.OpcUaStatus.BadInternalError);
|
||||
AwaitAssert(() => capture.Count.ShouldBe(3), PresenceBudget);
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
capture.Count.ShouldBe(3, "7 of 10 failures were suppressed from the cluster-wide log");
|
||||
|
||||
now = now.AddMinutes(1);
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 99, DateTime.UtcNow));
|
||||
|
||||
AwaitAssert(() => capture.Count.ShouldBe(5), PresenceBudget);
|
||||
var entries = capture.ToArray();
|
||||
entries[3].Message.ShouldContain("7");
|
||||
entries[3].Message.ShouldContain("suppressed");
|
||||
entries[3].VirtualTagId.ShouldBe("vt-chatty");
|
||||
entries[4].Message.ShouldBe("boom");
|
||||
}
|
||||
|
||||
private sealed class FailingEvaluator(string reason) : IVirtualTagEvaluator
|
||||
{
|
||||
public VirtualTagEvalResult Evaluate(string id, string expr, IReadOnlyDictionary<string, object?> deps)
|
||||
=> VirtualTagEvalResult.Failure(reason);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user