From 4d3fc9c8167b075fbaee1bfa0121ef6070efba1c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 10:27:49 -0400 Subject: [PATCH] fix(runtime): restart-safe alarm host, mux last-value replay, VT log rate limit, Calculation remediation (E4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .../Validation/DraftValidator.cs | 48 +++++ .../CalculationDriver.cs | 48 ++++- .../CalculationEvaluator.cs | 39 +++- .../DependencyMuxTagUpstreamSource.cs | 28 ++- .../ScriptedAlarms/ScriptedAlarmHostActor.cs | 161 ++++++++++++++--- .../DependencyConsumerMuxAdapter.cs | 5 +- .../VirtualTags/DependencyMuxActor.cs | 68 ++++--- .../VirtualTags/VirtualTagActor.cs | 36 +++- .../DraftValidatorCalculationTests.cs | 56 +++++- .../CalculationDriverRemediationTests.cs | 166 ++++++++++++++++++ .../CalculationDriverTests.cs | 6 +- .../DependencyMuxTagUpstreamSourceTests.cs | 22 +++ .../ScriptedAlarmHostActorRestartTests.cs | 145 +++++++++++++++ .../VirtualTags/DependencyMuxReplayTests.cs | 101 +++++++++++ ...VirtualTagActorFailureLogRateLimitTests.cs | 55 ++++++ 15 files changed, 921 insertions(+), 63 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverRemediationTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorRestartTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorFailureLogRateLimitTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs index 04e08fe9..3b8d0bb1 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs @@ -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)); + } + } + + /// Read the trigger fields from a calc tag's TagConfig: changeTriggered + /// (default true, the driver's own default) and timerIntervalMs (null when absent or not a + /// number). Never throws — malformed JSON yields the defaults. + 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); + } } /// Extract the scriptId string from a calc tag's schemaless TagConfig JSON. Never diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs index dc0d8ba1..905fff13 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs @@ -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 /// 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); } + /// Value equality for the dedup: structural for arrays (R07:H8 — object.Equals is reference + /// equality for a double[], so an array-returning timer script republished every tick through + /// the dual-namespace fan-out and the historian), ordinary + /// otherwise. + 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); + } + /// Record that this tag's INPUTS are Bad, so no evaluation ran. Publishes the /// worst input's status — carrying the last known value — once per change of that status: a /// repeat of the same Bad is silent, while BadNoCommunication arriving over an earlier @@ -727,6 +755,12 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba /// Test/diagnostic accessor: number of timer ticks skipped for in-flight groups. internal long TimerSkippedTickCount => _timers?.SkippedTickCount ?? 0; + /// Test/diagnostic accessor: sources currently remembered as failing to compile. + internal int CompileFailureCountForTest => _evaluator.CompileFailureCount; + + /// Test/diagnostic accessor: compiled scripts currently held by the evaluator's cache. + internal int CompiledScriptCountForTest => _evaluator.CompiledScriptCount; + /// Test/diagnostic accessor: when no change-triggered evaluation is /// queued or in flight. A test that used to assert straight after — /// which evaluated inline — waits on this instead, so the assertion still runs against a settled @@ -770,6 +804,10 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba /// now carries the failing INPUT's sub-code rather than a fixed BadInternalError. public uint LastPublishedStatus; public DateTime LastTimestampUtc; + + /// 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. + public string? LastFailureReason; } private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs index 4296b993..ba4931c7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs @@ -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 _compileFailures = new(StringComparer.Ordinal); private bool _disposed; /// Initializes a new . @@ -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 } } + /// Record a compile failure for and log it ONCE — the next trigger + /// of the same source returns the remembered reason without a Roslyn round-trip or a second Warning. + 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))); + + /// Test/diagnostic: the number of sources currently remembered as failing to compile. + public int CompileFailureCount => _compileFailures.Count; + + /// Test/diagnostic: the number of compiled scripts currently cached. + public int CompiledScriptCount => _cache.Count; + /// 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(); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/DependencyMuxTagUpstreamSource.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/DependencyMuxTagUpstreamSource.cs index 9c7c741f..5466ab49 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/DependencyMuxTagUpstreamSource.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/DependencyMuxTagUpstreamSource.cs @@ -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 _cache = new(StringComparer.Ordinal); private readonly ConcurrentDictionary>> _observers @@ -91,6 +93,28 @@ public sealed class DependencyMuxTagUpstreamSource : ITagUpstreamSource } } + /// + /// Drop every cached snapshot whose path is not in . 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. + /// + /// The paths to keep. + /// The number of cached snapshots dropped. + public int RetainOnly(IReadOnlySet paths) + { + ArgumentNullException.ThrowIfNull(paths); + var dropped = 0; + foreach (var key in _cache.Keys) + { + if (!paths.Contains(key) && _cache.TryRemove(key, out _)) dropped++; + } + return dropped; + } + + /// Test/diagnostic: the number of paths currently retained. + public int RetainedPathCount => _cache.Count; + /// public DataValueSnapshot ReadTag(string path) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index ffa5dc4f..210a84ca 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -23,10 +23,16 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// cluster alerts DistributedPubSub topic. /// /// -/// The host is the engine's lifecycle owner: the caller (the driver-role startup) builds -/// the , constructs a -/// around that same upstream, and passes both here. The host disposes the engine in -/// . +/// The host is the engine's lifecycle owner. The restart-safe shape (R02:H1) is the +/// factory : +/// every incarnation builds its own + a +/// around it, and disposes them in . +/// Akka's default restart runs PostStop 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 +/// disposed engine and every later LoadAsync 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 +/// ); production should spawn through the factory. /// /// /// @@ -99,6 +105,9 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// Warning on every clean shutdown. private sealed record AlarmsLoadCanceled; + /// Default bound on one inbound alarm command's engine op (): 10 s. + 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 _onEngineEvent; + private readonly TimeSpan _commandTimeout; + + /// 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. + private readonly bool _ownsEngine; + + /// Set by so the it triggers can tell a restart + /// (the next incarnation reuses a legacy engine) from a genuine stop. + private bool _restarting; /// Cached local from the latest /// snapshot (null = unknown until the first snapshot arrives, or no wired). The @@ -152,6 +171,8 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// Optional node-local live-telemetry hub (Phase 5) each cluster-wide /// alerts transition is also emitted into; null (tests) skips the emit. /// The used to instantiate the actor. + /// 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. 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)); + /// + /// Restart-safe factory Props (R02:H1): each incarnation of the actor calls + /// in its constructor and owns what it returns — the engine must + /// be constructed around the returned upstream — so a supervision restart gets a fresh, + /// undisposed engine and the next loads cleanly. + /// + /// The OPC UA publish actor emissions are bridged to. + /// Optional dependency multiplexer the host registers interest with. + /// Builds a fresh (upstream, engine) pair; called once per incarnation. + /// The local cluster node id for the Primary alerts-emit gate; null ⇒ default-emit. + /// Test seam for the alerts-emit gate's driver-member count. + /// Optional node-local live-telemetry hub. + /// Bound on one inbound alarm command's engine op; null ⇒ . + /// The used to instantiate the actor. + public static Props Props( + IActorRef publishActor, + IActorRef? mux, + Func<(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine)> engineFactory, + NodeId? localNode = null, + Func? driverMemberCountProvider = null, + ITelemetryLocalHub? telemetryHub = null, + TimeSpan? commandTimeout = null) => + Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, engineFactory, localNode, driverMemberCountProvider, telemetryHub, commandTimeout)); + + /// Factory-built incarnation: owns (and on every stop disposes) the engine it builds. + /// The OPC UA publish actor emissions are bridged to. + /// Optional dependency multiplexer the host registers interest with. + /// Builds this incarnation's (upstream, engine) pair. + /// The local cluster node id for the Primary alerts-emit gate. + /// Test seam for the alerts-emit gate's driver-member count. + /// Optional node-local live-telemetry hub. + /// Bound on one inbound alarm command's engine op. + public ScriptedAlarmHostActor( + IActorRef publishActor, + IActorRef? mux, + Func<(DependencyMuxTagUpstreamSource Upstream, ScriptedAlarmEngine Engine)> engineFactory, + NodeId? localNode = null, + Func? 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; + } + /// Initializes a new instance of the class. /// The OPC UA publish actor emissions are bridged to. /// Optional dependency multiplexer the host registers dependency interest with. @@ -183,14 +258,29 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor NodeId? localNode = null, Func? 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? 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(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(); } + /// + /// Akka's default PreRestart stops the children and calls . The flag + /// lets that keep a legacy (Props-captured) engine alive for the next incarnation + /// instead of disposing the instance it is about to be re-created around. + protected override void PreRestart(Exception reason, object message) + { + _restarting = true; + base.PreRestart(reason, message); + } + /// 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(); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs index 7011492d..b8dc818f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs @@ -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); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs index fe57da4e..3af36d6a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs @@ -26,6 +26,20 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// 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. +/// +/// +/// Last-value retention + replay (R07:H6). For every ref that has EVER been registered the +/// mux retains the last (value, status, timestamp) 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 +/// BadWaitingForInitialData until each of its inputs happened to change again (a +/// scripted-alarm host restart has the same shape). A with +/// 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. +/// /// public sealed class DependencyMuxActor : ReceiveActor { @@ -33,14 +47,23 @@ public sealed class DependencyMuxActor : ReceiveActor /// 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. - public sealed record RegisterInterest(IReadOnlyList TagRefs, IActorRef Subscriber); + /// The refs the subscriber wants. + /// The subscriber. + /// When true, the mux immediately sends the subscriber one + /// holding the last retained value of every ref + /// in it has one for (nothing when it has none). + public sealed record RegisterInterest(IReadOnlyList TagRefs, IActorRef Subscriber, bool ReplayRetained = false); /// Unregister every interest held by . Sent on PostStop by /// the subscriber, or by Terminated handling when the mux watches. 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> _byRef = new(StringComparer.Ordinal); private readonly Dictionary> _bySubscriber = new(); + private readonly Dictionary _last = new(StringComparer.Ordinal); private readonly ILoggingAdapter _log = Context.GetLogger(); /// Creates props for the DependencyMuxActor. @@ -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(msg.TagRefs, StringComparer.Ordinal); _bySubscriber[msg.Subscriber] = refs; + List? 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 } } - /// Filters one batched driver production burst per item and delivers each subscriber's matched - /// slice as ONE . 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 evaluate its script once per burst instead of once per input. - /// The registered set is NOT just virtual-tag inputs: ContinuousHistorizationRecorder - /// registers interest in every historized ref, so on a historizing node can be - /// large and a sizeable fraction of each burst resolves. 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>(); foreach (var sub in subscribers) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs index 3d2a3058..63ea8736 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs @@ -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 _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 /// Optional reference to a dependency multiplexer actor. /// Optional node-local live-telemetry hub each script-log entry is also /// emitted into (Phase 5); null (tests) simply skips the emit. + /// Rate limiter for this tag's per-failure script-log entries; null ⇒ + /// the default 20 per minute + one summary. /// A configured for creating the actor. public static Props Props( string virtualTagId, @@ -113,7 +120,8 @@ public sealed class VirtualTagActor : ReceiveActor Func? publisherFactory = null, IReadOnlyList? 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(), mux, - telemetryHub)); + telemetryHub, + failureLogLimiter)); /// Initializes a virtual tag actor with the given configuration and dependencies. /// Unique identifier for the virtual tag. @@ -133,6 +142,8 @@ public sealed class VirtualTagActor : ReceiveActor /// Optional reference to a dependency multiplexer actor. /// Optional node-local live-telemetry hub each script-log entry is also /// emitted into (Phase 5); null (tests) simply skips the emit. + /// Rate limiter for this tag's per-failure script-log entries; null ⇒ + /// the default 20 per minute + one summary. public VirtualTagActor( string virtualTagId, string expression, @@ -141,7 +152,8 @@ public sealed class VirtualTagActor : ReceiveActor Func? publisherFactory, IReadOnlyList 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(OnDependencyChanged); Receive(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("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()) { diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorCalculationTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorCalculationTests.cs index 0b20f490..b542871f 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorCalculationTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorCalculationTests.cs @@ -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() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverRemediationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverRemediationTests.cs new file mode 100644 index 00000000..88015ffc --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverRemediationTests.cs @@ -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; + +/// +/// 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 +/// InitializeAsync so a Restart computes from current inputs (H6); array results dedup +/// structurally (H8); ShutdownAsync drops the compiled scripts (L2). +/// +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 Published, List Logs, CapturingLogger HostLog) + Build(params RawTagEntry[] tags) + { + var logs = new List(); + 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(); + 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 entries) : IScriptLogPublisher + { + public void Publish(ScriptLogEntry entry) { lock (entries) entries.Add(entry); } + } + + internal sealed class CapturingLogger : ILogger + { + public ConcurrentBag<(LogLevel Level, string Message)> Entries { get; } = new(); + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + => Entries.Add((logLevel, formatter(state, exception))); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs index 5eed4c6b..cf22038b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs @@ -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); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/DependencyMuxTagUpstreamSourceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/DependencyMuxTagUpstreamSourceTests.cs index d020d118..a3861648 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/DependencyMuxTagUpstreamSourceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/DependencyMuxTagUpstreamSourceTests.cs @@ -15,6 +15,28 @@ public sealed class DependencyMuxTagUpstreamSourceTests private static DataValueSnapshot Good(object? value, DateTime ts) => new(value, 0u, ts, ts); + /// 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. + [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(); + using var sub = src.SubscribeTag("old/renamed", (p, _) => seen.Add(p)); + + src.RetainOnly(new HashSet(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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorRestartTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorRestartTests.cs new file mode 100644 index 00000000..c4063d57 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorRestartTests.cs @@ -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; + +/// +/// R02:H1 — a supervision restart of used to dispose the +/// Props-captured engine and re-create the actor around the disposed instance, so every later +/// ApplyScriptedAlarms 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 +/// handler. Also R02:H3 — an inbound alarm command's engine op is bounded. +/// +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)); + } + + /// Apply, crash the handler (a null TagId makes Push throw → restart), apply again + /// and prove the second load registers interest and the alarm still activates. + private void RestartThenReload(IActorRef host, TestProbe publish, TestProbe mux) + { + host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan() })); + mux.ExpectMsg(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(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(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(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(Timeout); + host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow)); + publish.ExpectMsg(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(Timeout).State.Active.ShouldBeFalse(); + store.ObservedCancellation.ShouldBeTrue(); + } + + /// Store whose next SaveAsync blocks until its token is cancelled. + private sealed class HangingSaveStore : IAlarmStateStore + { + private readonly InMemoryAlarmStateStore _inner = new(); + public volatile bool HangNext; + public volatile bool ObservedCancellation; + + public Task LoadAsync(string alarmId, CancellationToken ct) => _inner.LoadAsync(alarmId, ct); + public Task> 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); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs new file mode 100644 index 00000000..ae42edb9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs @@ -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; + +/// +/// R07:H6 — the mux retains the last value per ever-registered ref and replays it to a registration +/// that asks (). A Calculation driver +/// Restart (new driver + new adapter) therefore republishes from current inputs instead of sitting +/// BadWaitingForInitialData until every input happens to change. +/// +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(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(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(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)); + } + + /// 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. + [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(); + 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(); + 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); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorFailureLogRateLimitTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorFailureLogRateLimitTests.cs new file mode 100644 index 00000000..74ee2e32 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorFailureLogRateLimitTests.cs @@ -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; + +/// R02:P6 — the per-failure cluster-wide is rate-limited per tag +/// (first N per window, then one summary); the Bad publish and the metric are unaffected. +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(); + 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(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 deps) + => VirtualTagEvalResult.Failure(reason); + } +}