diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index a51b2046..c4262df3 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -98,6 +98,7 @@ Virtual-tag nodes are materialised by `AddressSpaceApplier` and fed by `VirtualT - **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes the items of each `DriverInstanceActor.AttributeValueBatchPublished` to the `VirtualTagActor` instances that declared interest in those tag refs. Each `VirtualTagActor` holds the in-memory per-tag dependency map; the `IVirtualTagEvaluator` (`RoslynVirtualTagEvaluator`) receives the dependency snapshot through `EvaluateAsync`. Reads of never-pushed dependency refs return `null` values in the dependency snapshot. - **Evaluation is off the dispatcher thread (lmxopcua#779, #750 P0).** The actor used to call the blocking `Evaluate`, which parked an Akka dispatcher thread for the whole run — up to the 2 s budget per evaluation. It now calls `EvaluateAsync`: a passthrough, a compile failure or any synchronous evaluator completes in the same message turn as before, while a real script run (and a first-use compile, which is pushed to the pool) returns an incomplete task. Until it completes the actor **stashes** every further dependency / reassert message, then applies the result and unstashes in arrival order. So each trigger is still evaluated against the inputs as of that trigger, results are published in trigger order, and the dedup / Bad-transition state sees the same sequence — only the thread is freed. Pinned by `VirtualTagActorAsyncEvaluationTests` (a slow script on a one-thread dispatcher no longer starves a sibling actor; ordering, reassert and fault paths) and `RoslynVirtualTagEvaluatorAsyncTests`. +- **Timer-driven evaluation (lmxopcua#783).** `VirtualTag.TimerIntervalMs` re-evaluates the tag on a fixed cadence (minimum 50 ms, clamped) via an Akka scheduler owned by `VirtualTagActor` — started in `PreStart`, cancelled in `PostStop`, so a stop or a redeploy respawn never leaves a ticker. `ChangeTriggered=false` makes the timer the only trigger (dependencies are still tracked). A tick that lands while a script run is in flight is dropped rather than stashed, so a slow script never builds a backlog of runs. Before this the field was stored, validated and edited but never consumed, so a timer-only tag sat at `BadWaitingForInitialData` forever. The deploy gate `VirtualTagNoEffectiveTrigger` (mirror of `CalculationNoEffectiveTrigger`) rejects a tag with neither a timer nor a change-trigger with a literal `ctx.GetTag` read. Pinned by `VirtualTagActorTimerTests` and `DraftValidatorVirtualTagTriggerTests`. - **Calculation tags and scripted alarms still evaluate one script at a time per driver / per engine (deliberately, for now).** Neither blocks a thread — both await the run — but a slow script still delays its siblings. Running independent scripts in parallel was considered for #779 and not done, because it is not semantics-neutral: for Calculation, the single pump is what bounds a driver to one pool thread of script work (per-tag lanes would let N slow tags hold N pool threads, the pool Akka's dispatchers also run on, so it needs a bounded-parallelism knob), it would reorder publishes across tags, and the timer path would need per-tag exclusion with the pump; for scripted alarms, the engine gate is also the consistency boundary of the Part 9 state machine, persist-before-update and ack/shelve/load, a predicate's transition time is sampled just before its own run, and a store failure mid-batch today leaves later predicates un-run (in parallel they would already have run, moving their run-guard strike streaks and writing script logs for results that are then discarded). The out-of-process runner's batch-submit `IScriptRunner` (#750) is where that concurrency belongs, with the runner owning the bound. - **`IHistoryWriter`** — the equipment-namespace path threads `Historize` end-to-end and `VirtualTagHostActor` invokes the injected writer on historized results (H5); the writer is resolved through `DriverHostActor` DI with a `NullHistoryWriter` default. No *durable* writer ships on this seam — durable continuous historization now flows through the separate `ContinuousHistorizationRecorder` → HistorianGateway `WriteLiveValues` path (see the `IHistoryWriter` section above and [Historian.md](Historian.md)). 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 ef76b5da..c01c6675 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs @@ -40,6 +40,7 @@ public static class DraftValidator ValidateUnsEffectiveLeafUniqueness(draft, errors); ValidateEquipReferenceResolution(draft, errors); ValidateCalculationTags(draft, errors); + ValidateVirtualTagTriggers(draft, errors); ValidateSqlConnectionStringNotPersisted(draft, errors); errors.AddRange(SqlTagDeployChecks.UndeclaredQueryErrors(draft)); ValidateMTConnectSingleDevice(draft, errors); @@ -480,6 +481,32 @@ public static class DraftValidator } } + /// VirtualTag effective-trigger gate (lmxopcua#783), the VirtualTag mirror of + /// CalculationNoEffectiveTrigger: a VirtualTag evaluates on a literal ctx.GetTag dependency + /// changing (when ChangeTriggered) and/or on its TimerIntervalMs. Neither means it sits at + /// BadWaitingForInitialData forever while everything reports Healthy. + private static void ValidateVirtualTagTriggers(DraftSnapshot draft, List errors) + { + var scriptSourceById = draft.Scripts + .GroupBy(s => s.ScriptId, StringComparer.Ordinal) + .ToDictionary(g => g.Key, g => g.First().SourceCode, StringComparer.Ordinal); + foreach (var v in draft.VirtualTags) + { + if (v.TimerIntervalMs is > 0) continue; + var source = scriptSourceById.TryGetValue(v.ScriptId, out var s) ? s : null; + // An unresolved script is reported elsewhere; with no source there are no reads to count. + if (v.ChangeTriggered && source is null) continue; + if (v.ChangeTriggered && EquipmentScriptPaths.ExtractDependencyRefs(source).Count > 0) continue; + errors.Add(new("VirtualTagNoEffectiveTrigger", + v.ChangeTriggered + ? $"VirtualTag '{v.VirtualTagId}' 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." + : $"VirtualTag '{v.VirtualTagId}' has ChangeTriggered=false and no TimerIntervalMs — nothing would ever evaluate it. " + + "Enable the change trigger or add a timer.", + v.VirtualTagId)); + } + } + /// 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. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs index 01a4fd1f..dfd0de72 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs @@ -376,6 +376,8 @@ public sealed record EquipmentTagAlarmInfo(string AlarmType, int Severity, bool? /// Historize-only toggle is detected as a change. Defaults to false — matching both the CLR /// default of the bool VirtualTag.Historize column and the artifact-decode default when the flag is absent/non-bool — /// which keeps existing positional+named ctor call sites compiling and preserves byte-parity. +/// Evaluate on a dependency change (entity default true). +/// Periodic re-evaluation cadence in ms, or null for none (lmxopcua#783). public sealed record EquipmentVirtualTagPlan( string VirtualTagId, string EquipmentId, @@ -384,7 +386,9 @@ public sealed record EquipmentVirtualTagPlan( string DataType, string Expression, IReadOnlyList DependencyRefs, - bool Historize = false) + bool Historize = false, + bool ChangeTriggered = true, + int? TimerIntervalMs = null) { /// // Structural equality: the auto-generated record equality would compare DependencyRefs (an @@ -400,6 +404,8 @@ public sealed record EquipmentVirtualTagPlan( DataType == other.DataType && Expression == other.Expression && Historize == other.Historize && + ChangeTriggered == other.ChangeTriggered && + TimerIntervalMs == other.TimerIntervalMs && DependencyRefs.SequenceEqual(other.DependencyRefs, StringComparer.Ordinal); /// @@ -408,7 +414,7 @@ public sealed record EquipmentVirtualTagPlan( var hash = new HashCode(); hash.Add(VirtualTagId); hash.Add(EquipmentId); hash.Add(FolderPath); hash.Add(Name); hash.Add(DataType); hash.Add(Expression); - hash.Add(Historize); + hash.Add(Historize); hash.Add(ChangeTriggered); hash.Add(TimerIntervalMs); foreach (var r in DependencyRefs) hash.Add(r, StringComparer.Ordinal); return hash.ToHashCode(); } @@ -625,7 +631,9 @@ public static class AddressSpaceComposer DataType: v.DataType, Expression: expanded, DependencyRefs: EquipmentScriptPaths.ExtractDependencyRefs(expanded), - Historize: v.Historize); + Historize: v.Historize, + ChangeTriggered: v.ChangeTriggered, + TimerIntervalMs: v.TimerIntervalMs); }) .ToList(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs index 464a42c3..0de7dc7f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs @@ -1416,6 +1416,12 @@ internal sealed class ParsedDeploymentArtifact : IDisposable && (hEl.ValueKind == JsonValueKind.True || hEl.ValueKind == JsonValueKind.False) && hEl.GetBoolean(); + // Triggers (lmxopcua#783): entity defaults when absent/malformed — change-triggered, no timer. + var changeTriggered = !(el.TryGetProperty("ChangeTriggered", out var ctEl) + && ctEl.ValueKind == JsonValueKind.False); + int? timerMs = el.TryGetProperty("TimerIntervalMs", out var tmEl) + && tmEl.ValueKind == JsonValueKind.Number && tmEl.TryGetInt32(out var tmv) ? tmv : null; + // Substitute each {{equip}}/ with the owning equipment's backing RawPath BEFORE // extracting refs, so both Expression and DependencyRefs are machine-specific — byte-parity // with AddressSpaceComposer.Compose. @@ -1430,7 +1436,9 @@ internal sealed class ParsedDeploymentArtifact : IDisposable DataType: dataType ?? "BaseDataType", Expression: expanded, DependencyRefs: EquipmentScriptPaths.ExtractDependencyRefs(expanded), - Historize: historize)); + Historize: historize, + ChangeTriggered: changeTriggered, + TimerIntervalMs: timerMs)); } result.Sort((a, b) => 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 8fc2a2fe..0d9e0cf8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs @@ -43,6 +43,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash { + /// Smallest honoured timer cadence in ms — the same floor the DB check constraint, the AdminUI + /// guard and the Calculation driver enforce. A smaller configured value is clamped up to it. + public const int MinTimerIntervalMs = 50; + + /// Periodic re-evaluation tick (lmxopcua#783). Private to the actor; sent by the scheduler only + /// when the tag has a TimerIntervalMs. + private sealed class TimerTick { public static readonly TimerTick Instance = new(); private TimerTick() { } } + /// The asynchronous evaluation started by finished. Private to the actor. private sealed record EvaluationCompleted( Task Evaluation, uint WorstInputStatus, DateTime TimestampUtc); @@ -97,6 +105,9 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash private readonly IReadOnlyList _dependencyRefs; private readonly IActorRef? _mux; private readonly ITelemetryLocalHub? _telemetryHub; + private readonly bool _changeTriggered; + private readonly int? _timerIntervalMs; + private ICancelable? _timer; // 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). @@ -141,7 +152,9 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash IReadOnlyList? dependencyRefs = null, IActorRef? mux = null, ITelemetryLocalHub? telemetryHub = null, - ScriptLogRateLimiter? failureLogLimiter = null) => + ScriptLogRateLimiter? failureLogLimiter = null, + bool changeTriggered = true, + int? timerIntervalMs = null) => Akka.Actor.Props.Create(() => new VirtualTagActor( virtualTagId, expression, evaluator ?? NullVirtualTagEvaluator.Instance, @@ -150,7 +163,9 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash dependencyRefs ?? Array.Empty(), mux, telemetryHub, - failureLogLimiter)); + failureLogLimiter, + changeTriggered, + timerIntervalMs)); /// Initializes a virtual tag actor with the given configuration and dependencies. /// Unique identifier for the virtual tag. @@ -164,6 +179,10 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash /// 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. + /// Evaluate when a dependency changes (default). When false only the timer + /// evaluates; dependency values are still tracked. + /// Periodic re-evaluation cadence in ms (clamped to + /// ), or null for none (lmxopcua#783). public VirtualTagActor( string virtualTagId, string expression, @@ -173,8 +192,12 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash IReadOnlyList dependencyRefs, IActorRef? mux, ITelemetryLocalHub? telemetryHub = null, - ScriptLogRateLimiter? failureLogLimiter = null) + ScriptLogRateLimiter? failureLogLimiter = null, + bool changeTriggered = true, + int? timerIntervalMs = null) { + _changeTriggered = changeTriggered; + _timerIntervalMs = timerIntervalMs is > 0 ? Math.Max(timerIntervalMs.Value, MinTimerIntervalMs) : null; _virtualTagId = virtualTagId; _scriptId = scriptId; _expression = expression; @@ -188,6 +211,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash Receive(OnDependencyChanged); Receive(OnDependencyBatch); Receive(_ => OnReassertValue()); + Receive(_ => Evaluate(DateTime.UtcNow)); // Only reachable for a completion that outlived its behaviour (a supervision restart mid-evaluation // resets the behaviour stack): the incarnation it belonged to is gone, so it is dropped. Receive(_ => { }); @@ -196,6 +220,14 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash /// protected override void PreStart() { + if (_timerIntervalMs is { } ms) + { + // lmxopcua#783: timer-driven evaluation. Cancelled in PostStop, so a stop / redeploy respawn + // never leaves a stale ticker behind. + var period = TimeSpan.FromMilliseconds(ms); + _timer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(period, period, Self, TimerTick.Instance, Self); + } + if (_mux is not null && _dependencyRefs.Count > 0) { // ReplayRetained: a respawned child (plan edit, supervision restart) evaluates from the mux's @@ -207,6 +239,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash /// protected override void PostStop() { + _timer?.Cancel(); _mux?.Tell(new DependencyMuxActor.UnregisterInterest(Self)); } @@ -241,7 +274,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash private void OnDependencyChanged(DependencyValueChanged msg) { _dependencies[msg.TagId] = new DataValueSnapshot(msg.Value, msg.StatusCode, msg.TimestampUtc, msg.TimestampUtc); - Evaluate(msg.TimestampUtc); + if (_changeTriggered) Evaluate(msg.TimestampUtc); } /// Applies EVERY item in the batch to the dependency map first, then evaluates the script @@ -268,7 +301,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash new DataValueSnapshot(item.Value, item.StatusCode, item.TimestampUtc, item.TimestampUtc); } - Evaluate(items[count - 1].TimestampUtc); + if (_changeTriggered) Evaluate(items[count - 1].TimestampUtc); } /// Recomputes the expression against the current dependency map and emits/dedups the result. @@ -329,6 +362,9 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash ApplyEvaluation(msg.Evaluation, msg.WorstInputStatus, msg.TimestampUtc); Stash.UnstashAll(); }); + // A timer tick during a run is dropped, not stashed: the run in flight IS this interval's evaluation, + // and stashing would replay a backlog of ticks (one script run each) after a slow script (#783). + Receive(_ => { }); ReceiveAny(_ => Stash.Stash()); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs index f97fa387..e995b263 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs @@ -180,7 +180,9 @@ public sealed class VirtualTagHostActor : ReceiveActor publisherFactory: null, dependencyRefs: p.DependencyRefs, mux: _mux, - telemetryHub: _telemetryHub)); + telemetryHub: _telemetryHub, + changeTriggered: p.ChangeTriggered, + timerIntervalMs: p.TimerIntervalMs)); Context.Watch(child); _children[p.VirtualTagId] = child; newlySpawned.Add(p.VirtualTagId); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorVirtualTagTriggerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorVirtualTagTriggerTests.cs new file mode 100644 index 00000000..d6548a4d --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorVirtualTagTriggerTests.cs @@ -0,0 +1,57 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Validation; + +namespace ZB.MOM.WW.OtOpcUa.Configuration.Tests; + +/// lmxopcua#783 — VirtualTagNoEffectiveTrigger, the VirtualTag mirror of +/// CalculationNoEffectiveTrigger. +[Trait("Category", "Unit")] +public sealed class DraftValidatorVirtualTagTriggerTests +{ + private static Script ScriptRow(string id, string source) => new() + { + ScriptId = id, Name = id, SourceCode = source, SourceHash = "h-" + id, + }; + + private static VirtualTag Vt(bool changeTriggered, int? timer, string scriptId = "s1") => new() + { + VirtualTagId = "vt-1", EquipmentId = "eq-1", Name = "V", DataType = "Double", ScriptId = scriptId, + ChangeTriggered = changeTriggered, TimerIntervalMs = timer, + }; + + private static IEnumerable Run(VirtualTag vt, string source) => + DraftValidator.Validate(new DraftSnapshot + { + GenerationId = 1, ClusterId = "c", + VirtualTags = [vt], + Scripts = [ScriptRow("s1", source)], + }).Where(e => e.Code == "VirtualTagNoEffectiveTrigger"); + + [Fact] + public void Change_triggered_with_no_read_and_no_timer_is_rejected() => + Run(Vt(true, null), "return 1;").ShouldHaveSingleItem().Context.ShouldBe("vt-1"); + + [Fact] + public void Not_change_triggered_and_no_timer_is_rejected() => + Run(Vt(false, null), "return (double)ctx.GetTag(\"a/b\").Value;").ShouldHaveSingleItem(); + + [Fact] + public void A_timer_alone_is_an_effective_trigger() => + Run(Vt(true, 1000), "return 1;").ShouldBeEmpty(); + + [Fact] + public void A_literal_read_with_change_trigger_is_an_effective_trigger() => + Run(Vt(true, null), "return (double)ctx.GetTag(\"a/b\").Value;").ShouldBeEmpty(); + + [Fact] + public void An_unresolved_script_reports_no_trigger_error() + { + var errors = DraftValidator.Validate(new DraftSnapshot + { + GenerationId = 1, ClusterId = "c", VirtualTags = [Vt(true, null, "s-missing")], + }); + errors.ShouldNotContain(e => e.Code == "VirtualTagNoEffectiveTrigger"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DeploymentArtifactVirtualTagHistorizeParityTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DeploymentArtifactVirtualTagHistorizeParityTests.cs index a84da7c0..d6d28f79 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DeploymentArtifactVirtualTagHistorizeParityTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DeploymentArtifactVirtualTagHistorizeParityTests.cs @@ -136,6 +136,33 @@ public sealed class DeploymentArtifactVirtualTagHistorizeParityTests decoded.EquipmentVirtualTags.ShouldHaveSingleItem().Historize.ShouldBeFalse(); } + /// lmxopcua#783 — the triggers survive the artifact decode (they used to be dropped, so the runtime + /// never knew a timer existed); absent fields keep the entity defaults. + [Fact] + public void ParseComposition_carries_timer_and_change_trigger() + { + var blob = BlobOf(new + { + Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } }, + Scripts = new[] { new { ScriptId = "s-1", SourceCode = "return 1;" } }, + VirtualTags = new object[] + { + new { VirtualTagId = "vt-timer", EquipmentId = "eq-1", Name = "T", DataType = "Int32", ScriptId = "s-1", + ChangeTriggered = false, TimerIntervalMs = 1000 }, + new { VirtualTagId = "vt-default", EquipmentId = "eq-1", Name = "D", DataType = "Int32", ScriptId = "s-1" }, + }, + }); + + var decoded = DeploymentArtifact.ParseComposition(blob); + + var t = decoded.EquipmentVirtualTags.Single(p => p.VirtualTagId == "vt-timer"); + t.TimerIntervalMs.ShouldBe(1000); + t.ChangeTriggered.ShouldBeFalse(); + var d = decoded.EquipmentVirtualTags.Single(p => p.VirtualTagId == "vt-default"); + d.TimerIntervalMs.ShouldBeNull(); + d.ChangeTriggered.ShouldBeTrue(); + } + // The Pascal-case snapshot a VirtualTag EF entity serialises to (matches ConfigComposer's // DefaultIgnoreCondition.Never whole-entity serialisation — includes Historize). private static object ToSnapshot(VirtualTag v) => new diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorTimerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorTimerTests.cs new file mode 100644 index 00000000..935299b9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorTimerTests.cs @@ -0,0 +1,107 @@ +using Akka.Actor; +using Akka.Event; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags; + +/// lmxopcua#783 — VirtualTag.TimerIntervalMs drives evaluation: a timer-only tag (no +/// dependency) evaluates on its interval, ticks coalesce during a slow run, and the timer dies with the actor. +public sealed class VirtualTagActorTimerTests : RuntimeActorTestBase +{ + [Fact] + public void A_timer_only_tag_with_no_dependency_evaluates_on_its_interval() + { + var parent = CreateTestProbe(); + var evaluator = new CountingEvaluator(); + parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator, timerIntervalMs: 50)); + + var first = parent.ExpectMsg(TimeSpan.FromSeconds(5)); + var second = parent.ExpectMsg(TimeSpan.FromSeconds(5)); + ((int)second.Value!).ShouldBeGreaterThan((int)first.Value!); + } + + [Fact] + public void Without_a_timer_a_dependency_free_tag_never_evaluates() + { + var parent = CreateTestProbe(); + var evaluator = new CountingEvaluator(); + parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator)); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + evaluator.Calls.ShouldBe(0); + } + + [Fact] + public void ChangeTriggered_false_ignores_dependency_changes_and_evaluates_only_on_the_timer() + { + var parent = CreateTestProbe(); + var evaluator = new CountingEvaluator(); + var actor = parent.ChildActorOf(VirtualTagActor.Props( + "vt-1", "expr", evaluator: evaluator, changeTriggered: false, timerIntervalMs: 2000)); + actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow)); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(150)); + evaluator.Calls.ShouldBe(0); + parent.ExpectMsg(TimeSpan.FromSeconds(10)); + } + + [Fact] + public void Ticks_during_a_slow_run_coalesce_instead_of_piling_up() + { + var parent = CreateTestProbe(); + var evaluator = new CountingEvaluator(gateFirst: true); + parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator, timerIntervalMs: 50)); + + AwaitCondition(() => evaluator.Calls == 1, PresenceBudget); + Thread.Sleep(600); // ~12 ticks fire while the first run is in flight + evaluator.Calls.ShouldBe(1); + + evaluator.Release(); + parent.ExpectMsg(TimeSpan.FromSeconds(5)); + // A backlog replay would run ~11 more evaluations immediately; coalesced, at most one tick lands here. + Thread.Sleep(30); + evaluator.Calls.ShouldBeLessThanOrEqualTo(3); + } + + [Fact] + public void The_timer_stops_when_the_actor_stops() + { + var parent = CreateTestProbe(); + var evaluator = new CountingEvaluator(); + var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator, timerIntervalMs: 50)); + parent.ExpectMsg(TimeSpan.FromSeconds(5)); + + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + Thread.Sleep(100); // let any already-enqueued tick drain + // A scheduler that was never cancelled keeps Telling the dead actor: those land as dead letters. + Sys.EventStream.Subscribe(TestActor, typeof(DeadLetter)); + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); // ~12 intervals at TIMEFACTOR 3 + } + + private sealed class CountingEvaluator(bool gateFirst = false) : IVirtualTagEvaluator + { + private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _calls; + + public int Calls => Volatile.Read(ref _calls); + + public void Release() => _gate.TrySetResult(); + + public VirtualTagEvalResult Evaluate(string id, string expr, IReadOnlyDictionary deps) + => VirtualTagEvalResult.Ok(Interlocked.Increment(ref _calls)); + + public Task EvaluateAsync( + string id, string expr, IReadOnlyDictionary deps) + { + var n = Interlocked.Increment(ref _calls); + return gateFirst && n == 1 + ? _gate.Task.ContinueWith(_ => VirtualTagEvalResult.Ok(n), TaskScheduler.Default) + : Task.FromResult(VirtualTagEvalResult.Ok(n)); + } + } +}