diff --git a/docs/ScriptEditor.md b/docs/ScriptEditor.md index 88f7c54c..5321aca8 100644 --- a/docs/ScriptEditor.md +++ b/docs/ScriptEditor.md @@ -203,9 +203,9 @@ prefix. The concrete implementation `ScriptTagCatalog` reads `Tag` + literal against `DriverInstanceActor.AttributeValuePublished.FullReference`, which is the `FullName` field from `Tag.TagConfig` (see `ScriptTagCatalog.ExtractFullNameFromTagConfig` + `EquipmentNodeWalker.ExtractFullName`). -The UNS-path engine (`Core.VirtualTags.VirtualTagEngine`, keyed by the -slash-joined `Enterprise/Site/Area/Line/Equipment/TagName` browse path) is -dormant — it is not wired into the host — so UNS browse paths never resolve at +The former UNS-path cascade engine (`Core.VirtualTags.VirtualTagEngine`, keyed by the +slash-joined `Enterprise/Site/Area/Line/Equipment/TagName` browse path) was +retired (#603) — it was never wired into the host — so UNS browse paths never resolve at runtime and are intentionally NOT offered as completions. The catalog is scoped per Blazor circuit; each call creates and disposes its own @@ -233,8 +233,8 @@ deploy time. An unrelated `anything.GetTag("…")` no longer triggers it. **`SetVirtualTag` is a no-op in production.** The live single-tag evaluator (`RoslynVirtualTagEvaluator`) drops cross-tag `ctx.SetVirtualTag(...)` writes -(logged); the cascade `VirtualTagEngine` that honours them is dormant (not wired -into the host). So hover on a `ctx.SetVirtualTag("…")` path literal appends a +(logged); the cascade engine that would have honoured them (`VirtualTagEngine`) was retired +unwired (#603). So hover on a `ctx.SetVirtualTag("…")` path literal appends a note that the write is dropped in single-tag mode and will not take effect at runtime — the path completions still help author the literal, but the author is not misled into relying on a no-op. Making `SetVirtualTag` functional would diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index 20032d37..d58013ea 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -60,9 +60,7 @@ Parses the script source with `CSharpSyntaxTree.ParseText` (script kind), walks ## Virtual tag engine (`Core.VirtualTags`) -### `VirtualTagDefinition` - -One row per operator-authored tag. Fields: `Path` (UNS browse path; also the engine's internal id), `DataType` (`DriverDataType` enum; the evaluator coerces the script's return value to this and mismatch surfaces as `BadTypeMismatch`), `ScriptSource` (Roslyn C# script text), `ChangeTriggered` (re-evaluate on any input delta), `TimerInterval` (optional periodic cadence; null disables), `Historize` (route every evaluation result to `IHistoryWriter`). Change-trigger and timer are independent — a tag can be either, both, or neither. +> **Retired (#603):** the standalone cascade engine (`VirtualTagEngine`, `VirtualTagSource`, `VirtualTagDefinition`, `TimerTriggerScheduler`) had no production consumer and was deleted; virtual tags run through `VirtualTagHostActor` / `VirtualTagActor` + `RoslynVirtualTagEvaluator`. `Core.VirtualTags` now holds `DependencyGraph` (deploy-gate cycle check), `ITagUpstreamSource` and `IHistoryWriter`. Descriptions below of the engine's `Load`/cascade/timer behaviour are historical. ### `VirtualTagContext` @@ -77,56 +75,30 @@ Directed graph of tag paths. Edges run from a virtual tag to each path it reads. Cycle detection uses an **iterative** Tarjan's SCC implementation (no recursion, deep graphs cannot stack-overflow). Cycles of length > 1 and self-loops both reject; leaf references cannot form cycles with internal nodes. -### `VirtualTagEngine` lifecycle - -- **`Load(definitions)`** — clears prior state, compiles every script through `DependencyExtractor.Extract` + `ScriptEvaluator.Compile` (wrapped in `TimedScriptEvaluator`), registers each in `_tags` + `_graph`, runs `TopologicalSort` (cycle check), then for every upstream (non-virtual) path subscribes via `ITagUpstreamSource.SubscribeTag` and seeds `_valueCache` with `ReadTag`. Throws `InvalidOperationException` aggregating every compile failure at once so operators see the whole set; throws `DependencyCycleException` on cycles. Re-entrant — supports config-publish reloads by disposing the prior upstream subscriptions first. -- **`EvaluateAllAsync(ct)`** — evaluates every tag once in topological order. Called at startup so virtual tags have a defined initial value before subscriptions start. -- **`EvaluateOneAsync(path, ct)`** — single-tag evaluation. Entry point for `TimerTriggerScheduler` + tests. -- **`Read(path)`** — synchronous last-known-value lookup from `_valueCache`. Returns `BadNodeIdUnknown`-quality for unregistered paths. -- **`Subscribe(path, observer)`** — register a change observer; returns `IDisposable`. Does **not** emit a seed value. -- **`OnUpstreamChange(path, value)`** (internal, wired from the upstream subscription) — updates cache, notifies observers, launches `CascadeAsync` fire-and-forget so the driver's dispatcher isn't blocked. - -Evaluations are **serial across all tags** — `_evalGate` is a `SemaphoreSlim(1, 1)` held around every `EvaluateInternalAsync`. Parallelism is deferred (Phase 7 plan decision #19). Rationale: serial execution preserves the "earlier topological nodes computed before later dependents" invariant when two cascades race. Per-tag error isolation: a script exception or timeout sets that tag's quality to `BadInternalError` and logs a structured error; other tags keep evaluating. `OperationCanceledException` is re-thrown (shutdown path). - -Result coercion: `CoerceResult` maps the script's return value to the declared `DriverDataType` via `Convert.ToXxx`. Coercion failure returns null which the outer pipeline maps to `BadInternalError`; `BadTypeMismatch` is documented in the definition shape (`VirtualTagDefinition.DataType` doc) rather than emitted distinctly today. - -`IHistoryWriter.Record` fires per evaluation when `Historize = true`. The default `NullHistoryWriter` drops silently. - -### `TimerTriggerScheduler` - -Groups `VirtualTagDefinition`s by `TimerInterval`, one `System.Threading.Timer` per unique interval. Each tick evaluates the group's paths serially via `VirtualTagEngine.EvaluateOneAsync`. Errors per-tag log and continue. `Dispose()` cancels an internal `CancellationTokenSource` and disposes every timer. Independent of the change-trigger path — a tag with both triggers fires from both scheduling sources. - ### `ITagUpstreamSource` -What the engine pulls driver-tag values from. Reads are **synchronous** because user scripts call `ctx.GetTag(path)` inline — a blocking wire call per evaluation would kill throughput. Implementations are expected to serve from a last-known-value cache populated by subscription callbacks. The server's production implementation is `CachedTagUpstreamSource` (see Composition below). +What a value consumer (scripted alarms) pulls driver-tag values from. Reads are **synchronous** because user scripts call `ctx.GetTag(path)` inline — a blocking wire call per evaluation would kill throughput. Implementations are expected to serve from a last-known-value cache populated by subscription callbacks. The server's production implementation is `CachedTagUpstreamSource` (see Composition below). ### `IHistoryWriter` -Fire-and-forget sink for evaluation results when `VirtualTagDefinition.Historize = true`. Implementations must queue internally and drain on their own cadence — a slow historian must not block script evaluation. `NullHistoryWriter.Instance` is the no-op default. Scripted-alarm emissions flow through `Core.AlarmHistorian` via `HistorianAdapterActor` → the registered `IAlarmHistorianSink` (a separate concern; see [AlarmTracking.md](AlarmTracking.md)). *(The `Phase7EngineComposer.RouteToHistorianAsync` this used to name is gone — not renamed; the routing moved into the actor.)* +Fire-and-forget sink for evaluation results when a virtual tag is `Historize`d. Implementations must queue internally and drain on their own cadence — a slow historian must not block script evaluation. `NullHistoryWriter.Instance` is the no-op default. Scripted-alarm emissions flow through `Core.AlarmHistorian` via `HistorianAdapterActor` → the registered `IAlarmHistorianSink` (a separate concern; see [AlarmTracking.md](AlarmTracking.md)). *(The `Phase7EngineComposer.RouteToHistorianAsync` this used to name is gone — not renamed; the routing moved into the actor.)* **Equipment-namespace path (H5).** The `Historize` flag is threaded end-to-end on the equipment path: `VirtualTag.Historize` → composer + artifact-decode (byte-parity) → `EquipmentVirtualTagPlan.Historize` → `VirtualTagHostActor`, which calls `IHistoryWriter.Record(nodeId, snapshot)` for every historized result (in addition to publishing the live value). The writer is injectable via DI — `DriverHostActor` resolves `IHistoryWriter` (`TryAddSingleton`, `NullHistoryWriter` default) and threads it into `VirtualTagHostActor`. **This `IHistoryWriter` seam still ships no durable binding** (`NullHistoryWriter` default). Durable continuous historization of driver/virtual values is now handled by the separate `ContinuousHistorizationRecorder` (it taps the dependency-mux value fan-out → a crash-safe FasterLog outbox → the HistorianGateway's `WriteLiveValues` path; see [Historian.md](Historian.md)), not through this seam. A deployment can still bind a custom `IHistoryWriter` via DI. ## Dispatch integration -Per [ADR-002](v2/implementation/adr-002-driver-vs-virtual-dispatch.md) Option B, there is a single `GenericDriverNodeManager`. `VirtualTagSource` implements `IReadable` + `ISubscribable` over a `VirtualTagEngine`: - -- `ReadAsync` fans each path through `engine.Read(...)`. -- `SubscribeAsync` calls `engine.Subscribe` per path and forwards each engine observer callback as an `OnDataChange` event; emits an initial-data callback per OPC UA convention. -- `UnsubscribeAsync` disposes every per-path engine subscription it holds. -- **`IWritable` is deliberately not implemented.** Virtual-tag nodes are not client-writable because `OtOpcUaNodeManager.EnsureVariable` materialises every SDK variable with `AccessLevel = AccessLevels.CurrentRead`; the SDK base `CustomNodeManager2.Write` returns `BadNotWritable` for read-only nodes and v2 has no client-write dispatch path. Scripts are the only write path via `ctx.SetVirtualTag`. - -`NodeSourceKind` on each `DriverAttributeInfo` (set by `EquipmentNodeWalker` at address-space build time) drives which backend handles a read. See [ReadWriteOperations.md](ReadWriteOperations.md) and [v1/Subscriptions.md](v1/Subscriptions.md) for the broader dispatch framing. +Virtual-tag nodes are materialised by `AddressSpaceApplier` and fed by `VirtualTagActor` publishes; they are not client-writable (`AccessLevel = CurrentRead`), scripts being the only write path via `ctx.SetVirtualTag`. See [ReadWriteOperations.md](ReadWriteOperations.md). ## Upstream reads + history -`ITagUpstreamSource` and `IHistoryWriter` are the two ports the engine requires from its host. Both live in `Core.VirtualTags`. In the v2 actor system: +`ITagUpstreamSource` and `IHistoryWriter` are the two ports value consumers require from their host. Both live in `Core.VirtualTags`. In the v2 actor system: - **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes `DriverInstanceActor.AttributeValuePublished` events 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 synchronously on the actor message thread. Reads of never-pushed dependency refs return `null` values in the dependency snapshot. -- **`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. The standalone `VirtualTagEngine` likewise receives `NullHistoryWriter` by 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)). +- **`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)). ## Composition -`AddressSpaceComposer` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs`) is a pure static function that flattens config-DB entities into an `AddressSpaceComposition` value (UNS topology + raw subtree + driver-instance plans + scripted-alarm plans). `AddressSpaceApplier` applies that result into the OPC UA SDK node manager. Neither class has knowledge of `VirtualTagEngine` or `ScriptedAlarmEngine`. +`AddressSpaceComposer` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs`) is a pure static function that flattens config-DB entities into an `AddressSpaceComposition` value (UNS topology + raw subtree + driver-instance plans + scripted-alarm plans). `AddressSpaceApplier` applies that result into the OPC UA SDK node manager. Neither class has knowledge of `ScriptedAlarmEngine`. In the v2 actor system, virtual-tag engine composition is owned by the driver-role host actor tree: @@ -134,7 +106,7 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro - `RoslynVirtualTagEvaluator` (`src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs`) is injected into each `VirtualTagActor` as its `IVirtualTagEvaluator`. It holds a per-source `CompiledScriptCache` keyed by script source and compiles on first use. - `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) receives every `DriverInstanceActor.AttributeValuePublished` event and routes it to the `VirtualTagActor` instances that registered interest in that tag ref. -`VirtualTagEngine`, `VirtualTagSource`, `TimerTriggerScheduler`, and `ITagUpstreamSource` are available as standalone Core.VirtualTags primitives and remain the correct composition path for non-actor deployments (integration tests, future standalone runtimes). +`ITagUpstreamSource` and `IHistoryWriter` remain available as `Core.VirtualTags` ports. ## Key source files @@ -148,14 +120,10 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/DependencyExtractor.cs` — static `ctx.GetTag` / `ctx.SetVirtualTag` inference - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLoggerFactory.cs` — per-script Serilog logger - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs` — error mirror to main log -- `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagDefinition.cs` — per-tag config record - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagContext.cs` — evaluation-scoped `ctx` - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/DependencyGraph.cs` — Kahn topo-sort + iterative Tarjan SCC -- `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagEngine.cs` — load / evaluate / cascade pipeline -- `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/TimerTriggerScheduler.cs` — periodic re-evaluation - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs` — driver-tag read + subscribe port - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/IHistoryWriter.cs` — historize sink port + `NullHistoryWriter` -- `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagSource.cs` — `IReadable` + `ISubscribable` adapter - `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs` — actor that receives `DependencyValueChanged` from the mux and invokes `IVirtualTagEvaluator` per expression - `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs` — routes `DriverInstanceActor.AttributeValuePublished` to interested `VirtualTagActor` subscribers - `src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs` — production `IVirtualTagEvaluator` binding; holds a per-source `CompiledScriptCache` diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs index 0dc8825a..244b78f9 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs @@ -5,8 +5,7 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Engines; /// /// Abstraction over the compiled virtual-tag expression engine. Runtime consumes this so /// can stay free of Roslyn / scripting machinery and the -/// production wiring binds an adapter over VirtualTagEngine from -/// Core.VirtualTags. +/// production wiring binds RoslynVirtualTagEvaluator (Host). /// public interface IVirtualTagEvaluator { @@ -98,7 +97,7 @@ public static class ScriptEvalFailureReasons } /// Returns from every call. Bound by default -/// when the production VirtualTagEngine adapter hasn't been registered (Mac dev, tests). +/// when the production RoslynVirtualTagEvaluator hasn't been registered (Mac dev, tests). public sealed class NullVirtualTagEvaluator : IVirtualTagEvaluator { public static readonly NullVirtualTagEvaluator Instance = new(); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverAttributeInfo.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverAttributeInfo.cs index 350a9037..03455f0a 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverAttributeInfo.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverAttributeInfo.cs @@ -39,7 +39,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; /// /// /// Set when is — stable -/// logical id the VirtualTagEngine addresses by. Null otherwise. +/// logical id the virtual-tag runtime addresses by. Null otherwise. /// /// /// Set when is — @@ -78,7 +78,7 @@ public sealed record DriverAttributeInfo( /// Discriminates which runtime subsystem owns this node's Read/Write/ /// Subscribe dispatch. Driver = a real IDriver capability surface; /// Virtual = a Phase 7 .VirtualTagId'd tag -/// computed by the VirtualTagEngine; ScriptedAlarm = a scripted Part 9 alarm +/// computed by the virtual-tag runtime; ScriptedAlarm = a scripted Part 9 alarm /// materialized by the ScriptedAlarmEngine. /// public enum NodeSourceKind diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/VirtualTagContext.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/VirtualTagContext.cs index fbfae77c..52ce1752 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/VirtualTagContext.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/VirtualTagContext.cs @@ -8,9 +8,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; /// Per-evaluation for a virtual-tag script. Reads come /// out of the engine's last-known-value cache (driver tags updated via the /// ITagUpstreamSource subscription, virtual tags updated by prior -/// evaluations). Writes route through VirtualTagEngine's -/// OnScriptSetVirtualTag callback so cross-tag write side effects still -/// participate in change-trigger cascades (via the engine's CascadeAsync). +/// evaluations). Writes route through the host-supplied set-virtual-tag callback. /// /// /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs index 42cf1790..1ade6dd0 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs @@ -81,10 +81,12 @@ public static class ScriptSandbox // surface (operator-authored hosting helpers, third-party packages, app code) // is not on the list and stays unreachable. var references = new List(); + // #603: both halves are cached. CreateFromFile reads + maps the DLL, and the BCL half is ~170 + // assemblies, so doing it per compile was the dominant fixed cost of every ScriptEvaluator build. + // MetadataReference is immutable and thread-safe, so sharing instances across compilations is safe. foreach (var asm in pinnedAssemblies) - references.Add(MetadataReference.CreateFromFile(asm.Location)); - foreach (var path in EnumerateBclAssemblyPaths()) - references.Add(MetadataReference.CreateFromFile(path)); + references.Add(PinnedReferences.GetOrAdd(asm, static a => MetadataReference.CreateFromFile(a.Location))); + references.AddRange(BclReferences.Value); var imports = new[] { @@ -97,6 +99,13 @@ public static class ScriptSandbox return new SandboxConfig(references, imports); } + private static readonly System.Collections.Concurrent.ConcurrentDictionary PinnedReferences = new(); + + // The trusted-platform set is process-constant, so the BCL reference list is context-independent. + private static readonly Lazy> BclReferences = new( + static () => EnumerateBclAssemblyPaths().Select(static p => (MetadataReference)MetadataReference.CreateFromFile(p)).ToArray(), + LazyThreadSafetyMode.ExecutionAndPublication); + private static IEnumerable EnumerateBclAssemblyPaths() { // The .NET host advertises the resolved runtime-shared-framework + BCL DLL set diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs index 3616abe0..0edfd654 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs @@ -15,11 +15,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; /// from a last-known-value cache populated by the subscription callbacks. /// /// -/// The subscription path feeds 's -/// OnUpstreamChange callback, which updates the engine's value cache and -/// schedules CascadeAsync to re-evaluate every change-driven dependent in -/// topological order. One subscription per distinct upstream tag path; the -/// engine tracks the mapping itself. +/// The subscription path feeds the consumer's change callback (scripted alarms via +/// DependencyMuxTagUpstreamSource). One subscription per distinct upstream tag path. /// /// public interface ITagUpstreamSource diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/TimerTriggerScheduler.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/TimerTriggerScheduler.cs deleted file mode 100644 index 93b1f5ed..00000000 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/TimerTriggerScheduler.cs +++ /dev/null @@ -1,162 +0,0 @@ -using Serilog; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -/// -/// Periodic re-evaluation scheduler for tags with a non-null -/// . Independent of the -/// change-trigger path — a tag can be timer-only, change-only, or both. One -/// per interval-group keeps the wire count -/// low regardless of tag count. -/// -/// -/// -/// Each timer group carries a per-group in-flight flag (see -/// TickGroup.InFlight). When the timer fires while a tick for the same -/// group is still running, the new callback skips the work and increments -/// rather than blocking a thread-pool thread on -/// the engine's evaluation gate. This bounds the work outstanding at one tick -/// per group, regardless of how long an individual evaluation takes. -/// -/// -public sealed class TimerTriggerScheduler : IDisposable -{ - private readonly VirtualTagEngine _engine; - private readonly ILogger _logger; - private readonly List _timers = []; - private readonly List _groups = []; - private readonly CancellationTokenSource _cts = new(); - private long _skippedTickCount; - private bool _disposed; - private bool _started; - - /// Initializes a new instance of the class. - /// The virtual tag engine to trigger evaluations on. - /// A logger for diagnostic output. - public TimerTriggerScheduler(VirtualTagEngine engine, ILogger logger) - { - _engine = engine ?? throw new ArgumentNullException(nameof(engine)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - } - - /// - /// Diagnostic counter: number of timer callbacks that skipped their work because - /// the prior tick for the same group was still running. Exposed for tests + - /// operational metrics. Monotonic; never resets. - /// - public long SkippedTickCount => Interlocked.Read(ref _skippedTickCount); - - /// - /// Stand up one per unique interval. All tags with - /// matching interval share a timer; each tick triggers re-evaluation of the - /// group in topological order so cascades are consistent with change-triggered - /// behavior. - /// - /// The virtual tag definitions to schedule timers for. - /// Thrown when Start is called more than once on the same instance. - public void Start(IReadOnlyList definitions) - { - if (_disposed) throw new ObjectDisposedException(nameof(TimerTriggerScheduler)); - if (_started) throw new InvalidOperationException( - "TimerTriggerScheduler.Start has already been called. Create a new instance for each Load cycle."); - _started = true; - - var byInterval = definitions - .Where(d => d.TimerInterval.HasValue && d.TimerInterval.Value > TimeSpan.Zero) - .GroupBy(d => d.TimerInterval!.Value); - - foreach (var group in byInterval) - { - var paths = group.Select(d => d.Path).ToArray(); - var interval = group.Key; - var ctx = new TickGroup(paths); - _groups.Add(ctx); - var timer = new Timer(_ => OnTimer(ctx), null, interval, interval); - _timers.Add(timer); - _logger.Information("TimerTriggerScheduler: {TagCount} tag(s) on {Interval} cadence", - paths.Length, interval); - } - } - - private void OnTimer(TickGroup ctx) - { - if (_cts.IsCancellationRequested) return; - - // Skip the tick when the prior one for this group is still running. Without - // this guard a slow evaluation (or one waiting on the engine's _evalGate) would - // cause subsequent timer callbacks to each pin a thread-pool thread on the - // gate, compounding under high tick rates. - if (Interlocked.CompareExchange(ref ctx.InFlight, 1, 0) != 0) - { - Interlocked.Increment(ref _skippedTickCount); - return; - } - - // Run async without blocking the timer's pool-thread callback. The task is - // fire-and-forget — failures are logged inside RunTickAsync; the InFlight flag - // is reset in the finally block so the next tick can proceed. - _ = RunTickAsync(ctx); - } - - private async Task RunTickAsync(TickGroup ctx) - { - try - { - foreach (var p in ctx.Paths) - { - if (_cts.IsCancellationRequested) return; - try - { - await _engine.EvaluateOneAsync(p, _cts.Token).ConfigureAwait(false); - } - catch (OperationCanceledException) - { - return; - } - catch (Exception ex) - { - _logger.Error(ex, "TimerTriggerScheduler evaluate failed for {Path}", p); - } - } - } - finally - { - Interlocked.Exchange(ref ctx.InFlight, 0); - } - } - - /// Releases all timers and disposes the scheduler's resources. - public void Dispose() - { - if (_disposed) return; - _disposed = true; - _cts.Cancel(); - foreach (var t in _timers) - { - try { t.Dispose(); } catch { } - } - _timers.Clear(); - _groups.Clear(); - _cts.Dispose(); - } - - /// Groups tags that share the same timer interval. - private sealed class TickGroup - { - // 0 = idle, 1 = a tick is currently running (or queued) for this group. Use - // Interlocked.CompareExchange so a timer callback observes a consistent "is the - // prior tick still running" answer without taking a lock. - /// Flag indicating whether a tick is currently in flight for this group. - public int InFlight; - - /// Gets the list of tag paths in this group. - public IReadOnlyList Paths { get; } - - /// Initializes a new instance of the class. - /// The tag paths in this timer group. - public TickGroup(IReadOnlyList paths) - { - Paths = paths; - } - } -} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagDefinition.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagDefinition.cs deleted file mode 100644 index 60b7f953..00000000 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagDefinition.cs +++ /dev/null @@ -1,41 +0,0 @@ -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -/// -/// Operator-authored virtual-tag configuration row. Phase 7 Stream E (config DB -/// schema) materializes these from the VirtualTag + Script tables on -/// publish; the engine ingests a list of them at load time. -/// -/// -/// UNS tag path — Enterprise/Site/Area/Line/Equipment/TagName. Used both as -/// the engine's internal id and the OPC UA browse path. -/// -/// -/// Expected return type. The evaluator coerces the script's return value to this -/// type before publishing; mismatch surfaces as BadTypeMismatch quality on -/// the tag. -/// -/// Roslyn C# script source. Must compile under ScriptSandbox. -/// -/// True if any input tag's change (value / status / timestamp delta) should trigger -/// re-evaluation. Operator picks per tag — usually true for inputs that change at -/// protocol rates. -/// -/// -/// Optional periodic re-evaluation cadence. Null = timer-driven disabled. Both can -/// be enabled simultaneously; independent scheduling paths both end at -/// 's EvaluateInternalAsync. -/// -/// -/// When true, every evaluation result is forwarded to the configured -/// . Operator-set per tag; the Admin UI exposes as a -/// checkbox. -/// -public sealed record VirtualTagDefinition( - string Path, - DriverDataType DataType, - string ScriptSource, - bool ChangeTriggered = true, - TimeSpan? TimerInterval = null, - bool Historize = false); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagEngine.cs deleted file mode 100644 index 7009684c..00000000 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagEngine.cs +++ /dev/null @@ -1,568 +0,0 @@ -using System.Collections.Concurrent; -using Serilog; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Core.Scripting; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -/// -/// The Phase 7 virtual-tag evaluation engine. Ingests a set of -/// s at load time, compiles each script against -/// , builds the dependency graph, subscribes to every -/// referenced upstream tag, and schedules re-evaluations on change + on timer. -/// -public sealed class VirtualTagEngine : IDisposable -{ - private readonly ITagUpstreamSource _upstream; - private readonly IHistoryWriter _history; - private readonly ScriptLoggerFactory _loggerFactory; - private readonly ILogger _engineLogger; - private readonly Func _clock; - private readonly TimeSpan _scriptTimeout; - - private readonly DependencyGraph _graph = new(); - private readonly Dictionary _tags = new(StringComparer.Ordinal); - - /// - /// Compile cache for every virtual-tag script. Routes 's - /// calls through the - /// cache so the collectible - /// each compile produces is actually disposed on the publish-replace path: the - /// cache's disposes - /// every materialised evaluator before dropping its dictionary entry, so a - /// config-publish releases the prior generation's ALCs and the per-publish - /// accretion is actually freed in production. Pre-fix, the engine called - /// ScriptEvaluator.Compile directly, which left the ALCs rooted until - /// the process exited. - /// - private readonly CompiledScriptCache _compileCache = new(); - - private readonly ConcurrentDictionary _valueCache = new(StringComparer.Ordinal); - private readonly ConcurrentDictionary>> _observers - = new(StringComparer.Ordinal); - private readonly List _upstreamSubscriptions = []; - private readonly SemaphoreSlim _evalGate = new(1, 1); - private bool _loaded; - private bool _disposed; - - /// - /// Initializes a new instance of the class. - /// - /// The upstream tag source for reading input tags. - /// Factory for creating script loggers. - /// Logger for engine events and errors. - /// Optional history writer for recording tag evaluations; defaults to no-op. - /// Optional function providing current UTC time; defaults to . - /// Optional timeout for script evaluation; defaults to . - public VirtualTagEngine( - ITagUpstreamSource upstream, - ScriptLoggerFactory loggerFactory, - ILogger engineLogger, - IHistoryWriter? historyWriter = null, - Func? clock = null, - TimeSpan? scriptTimeout = null) - { - _upstream = upstream ?? throw new ArgumentNullException(nameof(upstream)); - _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory)); - _engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger)); - _history = historyWriter ?? NullHistoryWriter.Instance; - _clock = clock ?? (() => DateTime.UtcNow); - _scriptTimeout = scriptTimeout ?? TimedScriptEvaluator.DefaultTimeout; - } - - /// Registered tag paths, in topological order. Empty before . - public IReadOnlyCollection LoadedTagPaths => _tags.Keys; - - /// Compile + register every tag in . Throws on cycle or any compile failure. - /// List of virtual tag definitions to load and compile. - public void Load(IReadOnlyList definitions) - { - if (_disposed) throw new ObjectDisposedException(nameof(VirtualTagEngine)); - if (definitions is null) throw new ArgumentNullException(nameof(definitions)); - - // Start from a clean slate — supports config-publish reloads. - UnsubscribeFromUpstream(); - _tags.Clear(); - _graph.Clear(); - _compileCache.Clear(); - - var compileFailures = new List(); - var seenPaths = new HashSet(StringComparer.Ordinal); - foreach (var def in definitions) - { - if (!seenPaths.Add(def.Path)) - { - compileFailures.Add($"{def.Path}: duplicate path — only one definition per path is allowed"); - continue; - } - - if (!IsSupportedDataType(def.DataType)) - { - compileFailures.Add( - $"{def.Path}: unsupported DataType DriverDataType.{def.DataType} — virtual tags only support scalar primitive types"); - continue; - } - - try - { - var extraction = DependencyExtractor.Extract(def.ScriptSource); - if (!extraction.IsValid) - { - var msgs = string.Join("; ", extraction.Rejections.Select(r => r.Message)); - compileFailures.Add($"{def.Path}: dependency extraction rejected — {msgs}"); - continue; - } - - var evaluator = _compileCache.GetOrCompile(def.ScriptSource); - var timed = new TimedScriptEvaluator(evaluator, _scriptTimeout); - var scriptLogger = _loggerFactory.Create(def.Path); - - _tags[def.Path] = new VirtualTagState(def, timed, extraction.Reads, extraction.Writes, scriptLogger); - _graph.Add(def.Path, extraction.Reads); - } - catch (Exception ex) - { - compileFailures.Add($"{def.Path}: {ex.Message}"); - } - } - - // Validate every ctx.SetVirtualTag write target resolves to a registered virtual - // tag. A script writing to a non-existent virtual path would otherwise be silently - // dropped at runtime by OnScriptSetVirtualTag's warning-and-drop branch; catching - // it here surfaces operator typos as a publish failure. - foreach (var (path, state) in _tags) - { - foreach (var writeTarget in state.Writes) - { - if (!_tags.ContainsKey(writeTarget)) - { - compileFailures.Add( - $"{path}: ctx.SetVirtualTag target '{writeTarget}' is not a registered virtual tag"); - } - } - } - - if (compileFailures.Count > 0) - { - var joined = string.Join("\n ", compileFailures); - throw new InvalidOperationException( - $"Virtual-tag engine load failed. {compileFailures.Count} script(s) did not compile:\n {joined}"); - } - - // Cycle check — throws DependencyCycleException on offense. - _ = _graph.TopologicalSort(); - - // Subscribe to every referenced upstream path (driver tags only — virtual tags - // cascade internally). Seed the cache with current upstream values so first - // evaluations see something real. Iterate _tags.Values (the registered set) rather - // than definitions to avoid indexing by a raw input list that may contain duplicates. - var upstreamPaths = _tags.Values - .SelectMany(s => s.Reads) - .Where(p => !_tags.ContainsKey(p)) - .Distinct(StringComparer.Ordinal); - foreach (var path in upstreamPaths) - { - _valueCache[path] = _upstream.ReadTag(path); - _upstreamSubscriptions.Add(_upstream.SubscribeTag(path, OnUpstreamChange)); - } - - _loaded = true; - _engineLogger.Information( - "VirtualTagEngine loaded {TagCount} tag(s), {UpstreamCount} upstream subscription(s)", - _tags.Count, _upstreamSubscriptions.Count); - } - - /// - /// Evaluate every registered tag once in topological order — used at startup so - /// virtual tags have a defined initial value rather than inheriting the cache - /// default. Also called after a config reload. - /// - /// Cancellation token to stop evaluation. - /// A task that represents the asynchronous operation. - public async Task EvaluateAllAsync(CancellationToken ct = default) - { - EnsureLoaded(); - var order = _graph.TopologicalSort(); - foreach (var path in order) - { - if (_tags.ContainsKey(path)) - await EvaluateOneAsync(path, ct).ConfigureAwait(false); - } - } - - /// Evaluate a single tag — used by the timer trigger + test hooks. - /// Path of the virtual tag to evaluate. - /// Cancellation token to stop evaluation. - /// A task that represents the asynchronous operation. - public Task EvaluateOneAsync(string path, CancellationToken ct = default) - { - EnsureLoaded(); - if (!_tags.ContainsKey(path)) - throw new ArgumentException($"Not a registered virtual tag: {path}", nameof(path)); - return EvaluateInternalAsync(path, ct); - } - - /// - /// Read the most recently evaluated value for . Driver - /// tags return the last-known upstream value; virtual tags return their last - /// evaluation result. - /// - /// Path of the tag to read. - /// The tag's most recently evaluated value, or a Bad-quality snapshot if the path is unrecognized. - public DataValueSnapshot Read(string path) - { - if (string.IsNullOrWhiteSpace(path)) - return new DataValueSnapshot(null, 0x80340000u, null, _clock()); - return _valueCache.TryGetValue(path, out var v) - ? v - : new DataValueSnapshot(null, 0x80340000u /* BadNodeIdUnknown */, null, _clock()); - } - - /// - /// Register an observer that fires on every evaluation of the given tag. - /// Returns an to unsubscribe. Does NOT fire a seed - /// value — subscribers call for the current value if needed. - /// - /// Path of the tag to subscribe to. - /// Callback invoked with the tag path and new value on each evaluation. - /// An that removes the observer when disposed. - public IDisposable Subscribe(string path, Action observer) - { - // Race-safe pattern paired with Unsub.Dispose: if Unsub.Dispose removed the - // dictionary entry between our GetOrAdd and the lock-protected Add, the list - // we hold a reference to is orphaned. Re-check the map under the lock and - // re-insert the list (or grab the current one) if needed, retrying until the - // dictionary observably contains the list we just added our observer to. - while (true) - { - var list = _observers.GetOrAdd(path, _ => []); - lock (list) - { - // Confirm the list is still the dictionary's value for this key. If - // Dispose removed the entry, _observers[path] either doesn't exist or - // points at a different (newer) list — retry. - if (_observers.TryGetValue(path, out var current) && ReferenceEquals(current, list)) - { - list.Add(observer); - return new Unsub(this, path, observer); - } - } - // Lost the race — Dispose pruned the list out from under us. Loop and - // either re-create or pick up the newer list. - } - } - - /// - /// Change-trigger entry point — called by the upstream subscription callback. - /// Updates the cache, fans out to observers (so OPC UA clients see the upstream - /// change too if they subscribed via the engine), and schedules every - /// change-triggered dependent for re-evaluation in topological order. - /// - /// Path of the upstream tag that changed. - /// New value of the tag. - internal void OnUpstreamChange(string path, DataValueSnapshot value) - { - _valueCache[path] = value; - NotifyObservers(path, value); - - // Fire-and-forget — the upstream subscription callback must not block the - // driver's dispatcher. Exceptions during cascade are handled per-tag inside - // EvaluateInternalAsync. - _ = CascadeAsync(path, CancellationToken.None); - } - - private async Task CascadeAsync(string upstreamPath, CancellationToken ct) - { - try - { - var dependents = _graph.TransitiveDependentsInOrder(upstreamPath); - foreach (var dep in dependents) - { - if (_tags.TryGetValue(dep, out var state) && state.Definition.ChangeTriggered) - await EvaluateInternalAsync(dep, ct).ConfigureAwait(false); - } - } - catch (Exception ex) - { - _engineLogger.Error(ex, "VirtualTagEngine cascade failed for upstream {Path}", upstreamPath); - } - } - - private async Task EvaluateInternalAsync(string path, CancellationToken ct) - { - if (!_tags.TryGetValue(path, out var state)) return; - - await _evalGate.WaitAsync(ct).ConfigureAwait(false); - try - { - var ctxCache = BuildReadCache(state.Reads); - - // Cold-start guard — when any upstream input is still unset or Bad-quality, - // publish a BadWaitingForInitialData snapshot so OPC UA clients see a defined - // quality rather than observing "not yet computed" as a stale Good value. - // Evaluating with nulls would throw inside the script (scripts cast - // ctx.GetTag(path).Value directly) and produce a persistent BadInternalError. - if (!AreInputsReady(ctxCache)) - { - var notReady = new DataValueSnapshot(null, 0x80320000u /* BadWaitingForInitialData */, null, _clock()); - _valueCache[path] = notReady; - NotifyObservers(path, notReady); - return; - } - - var context = new VirtualTagContext( - ctxCache, - (p, v) => OnScriptSetVirtualTag(p, v), - state.Logger, - _clock); - - DataValueSnapshot result; - try - { - var raw = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false); - var coerced = CoerceResult(raw, state.Definition.DataType); - // null from CoerceResult means the conversion threw (raw was non-null but - // not convertible to the declared type). Surface as BadInternalError so - // the OPC UA client sees a defined Bad quality rather than a Good null. - result = (raw is not null && coerced is null) - ? new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock()) - : new DataValueSnapshot(coerced, 0u, _clock(), _clock()); - } - catch (ScriptTimeoutException tex) - { - state.Logger.Warning("Script timed out after {Timeout}", tex.Timeout); - result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock()); - } - catch (OperationCanceledException) - { - throw; // shutdown path — don't misclassify - } - catch (Exception ex) - { - state.Logger.Error(ex, "Virtual-tag script threw"); - result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock()); - } - - _valueCache[path] = result; - NotifyObservers(path, result); - if (state.Definition.Historize) _history.Record(path, result); - } - finally - { - _evalGate.Release(); - } - } - - private IReadOnlyDictionary BuildReadCache(IReadOnlySet reads) - { - var map = new Dictionary(StringComparer.Ordinal); - foreach (var r in reads) - { - map[r] = _valueCache.TryGetValue(r, out var v) - ? v - : _upstream.ReadTag(r); - } - return map; - } - - /// - /// Returns true when every entry in has a - /// Good/Uncertain-quality (bit 31 = 0). - /// Readiness is gated on StatusCode alone — a Good-quality snapshot with a - /// null Value (e.g. a script that returns null, or a - /// ctx.SetVirtualTag call with a null argument) is considered ready and the - /// downstream script is allowed to run. The downstream script will observe the null - /// value and typically throw an NRE, which the outer evaluation catch maps to - /// BadInternalError — the correct sentinel for "script ran but faulted" - /// rather than "inputs not yet available". Previously the null-value check - /// caused a Good-quality null upstream to permanently block all its dependents at - /// BadWaitingForInitialData. - /// - private static bool AreInputsReady(IReadOnlyDictionary cache) - { - foreach (var kv in cache) - { - if (kv.Value is null) return false; - if ((kv.Value.StatusCode & 0x80000000u) != 0) return false; - } - return true; - } - - private void OnScriptSetVirtualTag(string path, object? value) - { - if (!_tags.ContainsKey(path)) - { - _engineLogger.Warning( - "Script attempted ctx.SetVirtualTag on non-virtual or non-registered path {Path}", path); - return; - } - var snap = new DataValueSnapshot(value, 0u, _clock(), _clock()); - _valueCache[path] = snap; - NotifyObservers(path, snap); - if (_tags[path].Definition.Historize) _history.Record(path, snap); - - // A cross-tag write must participate in the change-trigger cascade, exactly - // like an upstream delta — any change-triggered tag that reads this path - // would otherwise go stale until an unrelated trigger fires (see - // docs/VirtualTags.md, VirtualTagContext section). Fire-and-forget: this - // callback runs inside EvaluateInternalAsync with the non-reentrant - // _evalGate held, so the cascade must be scheduled, not invoked inline. - _ = CascadeAsync(path, CancellationToken.None); - } - - private void NotifyObservers(string path, DataValueSnapshot value) - { - if (!_observers.TryGetValue(path, out var list)) return; - Action[] snapshot; - lock (list) { snapshot = list.ToArray(); } - foreach (var obs in snapshot) - { - try { obs(path, value); } - catch (Exception ex) - { - _engineLogger.Warning(ex, "Virtual-tag observer for {Path} threw", path); - } - } - } - - private static object? CoerceResult(object? raw, DriverDataType target) - { - if (raw is null) return null; - try - { - return target switch - { - DriverDataType.Boolean => Convert.ToBoolean(raw), - DriverDataType.Int16 => Convert.ToInt16(raw), - DriverDataType.Int32 => Convert.ToInt32(raw), - DriverDataType.Int64 => Convert.ToInt64(raw), - DriverDataType.UInt16 => Convert.ToUInt16(raw), - DriverDataType.UInt32 => Convert.ToUInt32(raw), - DriverDataType.UInt64 => Convert.ToUInt64(raw), - DriverDataType.Float32 => Convert.ToSingle(raw), - DriverDataType.Float64 => Convert.ToDouble(raw), - DriverDataType.String => Convert.ToString(raw) ?? string.Empty, - DriverDataType.DateTime => raw is DateTime dt ? dt : Convert.ToDateTime(raw), - // Any DriverDataType not in the explicit list (currently Reference, or any - // future enum member added without coercion support) must NOT silently - // return the uncoerced raw value — that would surface as a wire-level - // type mismatch on the OPC UA Variant. Throwing here is caught by the - // outer catch and mapped to BadInternalError. Load-time validation in - // IsSupportedDataType ensures operators never publish such a tag. - _ => throw new InvalidOperationException( - $"Virtual-tag CoerceResult does not support DriverDataType.{target}"), - }; - } - catch - { - // Caller maps null to BadInternalError (0x80020000) — we let null propagate - // so the outer evaluation path sets the Bad quality on the snapshot. - return null; - } - } - - /// - /// The set of values can - /// honour. Definitions declaring any other type are rejected at - /// so an operator typo (or a future enum member added without coercion support) is - /// caught at publish time rather than silently producing a type-mismatched value. - /// - private static bool IsSupportedDataType(DriverDataType t) => t switch - { - DriverDataType.Boolean => true, - DriverDataType.Int16 => true, - DriverDataType.Int32 => true, - DriverDataType.Int64 => true, - DriverDataType.UInt16 => true, - DriverDataType.UInt32 => true, - DriverDataType.UInt64 => true, - DriverDataType.Float32 => true, - DriverDataType.Float64 => true, - DriverDataType.String => true, - DriverDataType.DateTime => true, - _ => false, - }; - - private void UnsubscribeFromUpstream() - { - foreach (var s in _upstreamSubscriptions) - { - try { s.Dispose(); } catch { /* best effort */ } - } - _upstreamSubscriptions.Clear(); - } - - private void EnsureLoaded() - { - if (!_loaded) throw new InvalidOperationException( - "VirtualTagEngine not loaded. Call Load(definitions) first."); - } - - /// - /// Releases all resources held by the engine, including upstream subscriptions and compiled script caches. - /// - public void Dispose() - { - if (_disposed) return; - _disposed = true; - UnsubscribeFromUpstream(); - _tags.Clear(); - _graph.Clear(); - _compileCache.Dispose(); - } - - /// - /// Gets the dependency graph for testing purposes. - /// - internal DependencyGraph GraphForTesting => _graph; - - private sealed class Unsub : IDisposable - { - private readonly VirtualTagEngine _engine; - private readonly string _path; - private readonly Action _observer; - /// - /// Initializes a new instance of the class. - /// - /// The parent virtual tag engine. - /// Path of the subscribed tag. - /// The observer callback to unsubscribe. - public Unsub(VirtualTagEngine e, string path, Action observer) - { - _engine = e; _path = path; _observer = observer; - } - /// - /// Removes the observer from the subscription list. - /// - public void Dispose() - { - if (_engine._observers.TryGetValue(_path, out var list)) - { - lock (list) - { - list.Remove(_observer); - // If we removed the last observer, prune the dictionary entry so a - // long-running server doesn't accumulate empty Lists for paths that - // saw transient subscriptions. The emptiness check is inside the same - // lock so a concurrent Subscribe can't slip an observer in after we - // observe the list as empty. - if (list.Count == 0) - { - // ICollection> removal is value-typed — only removes - // if both key + value still match (i.e. the dictionary still points - // at this list). This keeps a racing Subscribe's brand-new list safe. - ((ICollection>>>)_engine._observers) - .Remove(new KeyValuePair>>(_path, list)); - } - } - } - } - } - - internal sealed record VirtualTagState( - VirtualTagDefinition Definition, - TimedScriptEvaluator Evaluator, - IReadOnlySet Reads, - IReadOnlySet Writes, - ILogger Logger); -} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagSource.cs deleted file mode 100644 index e12ec270..00000000 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagSource.cs +++ /dev/null @@ -1,106 +0,0 @@ -using System.Collections.Concurrent; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -/// -/// Implements the driver-agnostic capability surface the -/// DriverNodeManager dispatches to when a node resolves to -/// NodeSource.Virtual. Reads return the engine's last-known -/// evaluation result; subscriptions forward engine-emitted change events as -/// events. -/// -/// -/// -/// is deliberately not implemented — OPC UA client -/// writes to virtual tags are rejected in DriverNodeManager before they -/// reach here. Scripts are the only write path, routed -/// through ctx.SetVirtualTag. -/// -/// -public sealed class VirtualTagSource : IReadable, ISubscribable -{ - private readonly VirtualTagEngine _engine; - private readonly ConcurrentDictionary _subs = new(StringComparer.Ordinal); - - /// - /// Initializes a new instance of the class. - /// - /// The virtual tag engine. - public VirtualTagSource(VirtualTagEngine engine) - { - _engine = engine ?? throw new ArgumentNullException(nameof(engine)); - } - - /// - /// Occurs when data values change. - /// - public event EventHandler? OnDataChange; - - /// - public Task> ReadAsync( - IReadOnlyList fullReferences, CancellationToken cancellationToken) - { - if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences)); - var results = new DataValueSnapshot[fullReferences.Count]; - for (var i = 0; i < fullReferences.Count; i++) - results[i] = _engine.Read(fullReferences[i]); - return Task.FromResult>(results); - } - - /// - public Task SubscribeAsync( - IReadOnlyList fullReferences, - TimeSpan publishingInterval, - CancellationToken cancellationToken) - { - if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences)); - - var handle = new SubscriptionHandle(Guid.NewGuid().ToString("N")); - var observers = new List(fullReferences.Count); - - // OPC UA convention: for each path, emit the initial-data callback BEFORE - // registering the change observer. This prevents a race where an upstream change - // fires the observer between the Subscribe call and the Read call, which would - // deliver a newer change event before the initial-data event, leaving the client - // with a stale last-known value. - foreach (var path in fullReferences) - { - var snap = _engine.Read(path); - OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, path, snap)); - observers.Add(_engine.Subscribe(path, (p, s) => - OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, p, s)))); - } - _subs[handle.DiagnosticId] = new Subscription(handle, observers); - - return Task.FromResult(handle); - } - - /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) - { - if (handle is null) throw new ArgumentNullException(nameof(handle)); - if (_subs.TryRemove(handle.DiagnosticId, out var sub)) - { - foreach (var d in sub.Observers) - { - try { d.Dispose(); } catch { } - } - } - return Task.CompletedTask; - } - - private sealed class SubscriptionHandle : ISubscriptionHandle - { - /// - /// Initializes a new instance of the class. - /// - /// The subscription identifier. - public SubscriptionHandle(string id) { DiagnosticId = id; } - - /// - public string DiagnosticId { get; } - } - - private sealed record Subscription(SubscriptionHandle Handle, IReadOnlyList Observers); -} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs index bf0bf55d..02e01074 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs @@ -3,12 +3,10 @@ using Microsoft.Extensions.Logging; namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation; /// -/// Periodic re-evaluation scheduler for calc tags carrying a timerIntervalMs. A deliberate -/// mirror of the VirtualTag TimerTriggerScheduler (Core.VirtualTags): one +/// Periodic re-evaluation scheduler for calc tags carrying a timerIntervalMs. One /// per distinct interval group (so the wire count stays low /// regardless of tag count) with a per-group in-flight flag that skips a tick when the prior tick for -/// the same group is still running. Revived inside the driver because the VT scheduler is welded -/// to VirtualTagEngine.EvaluateOneAsync; here each tick invokes the driver's own evaluate +/// the same group is still running. Each tick invokes the driver's own evaluate /// callback per tag in the group. /// internal sealed class CalculationTimerScheduler : IDisposable diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs index f759707f..7df9dd0e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs @@ -22,7 +22,7 @@ public sealed class LastValueCache { // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value has // been observed yet" (mirrored by CalculationDriver before its first evaluation, - // VirtualTagEngine for a freshly-materialised node, and OtOpcUaNodeManager / + // and OtOpcUaNodeManager / // AddressSpaceApplier for a just-deployed variable). MQTT is subscribe-first, so an unseen // RawPath is in exactly that state until its first publish arrives. Deliberately NOT // GoodNoData — that code is reserved (see NullHistorianDataSource, OtOpcUaNodeManager diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs index 663ef22c..ba69b0da 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs @@ -347,7 +347,7 @@ public sealed class ScriptAnalysisService : $"**Tag path** `{Code(info.Path)}`\n\n{info.Kind} · Type **{info.DataType}**" + (info.DriverInstanceId is null ? "" : $" · Driver `{Code(info.DriverInstanceId)}`"); // Truthfulness: ctx.SetVirtualTag(...) is a no-op in the live single-tag evaluator - // (RoslynVirtualTagEvaluator drops cross-tag writes; the cascade VirtualTagEngine is dormant). + // (RoslynVirtualTagEvaluator drops cross-tag writes). if (isSetVirtualTag) tagMd += "\n\n⚠ Cross-tag `SetVirtualTag` writes are currently dropped in single-tag mode " + "(the cascade engine is not wired into the host); this write will not take effect at runtime."; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs index f8a54971..a26ac5bb 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs @@ -40,7 +40,7 @@ public static class LocalDbRegistration public const string LocalDbSectionPath = "LocalDb"; /// Configuration section holding ReplicationOptions. - public const string ReplicationSectionPath = "LocalDb:Replication"; + public const string ReplicationSectionPath = Runtime.LocalDbReplicationPolicy.ReplicationSectionPath; /// /// Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 11b1c8d9..7bc8954a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -170,12 +170,9 @@ public static class LocalDbSetup /// true when this node is configured to replicate. public static bool ReplicationIsConfigured(IConfiguration configuration) { - ArgumentNullException.ThrowIfNull(configuration); - - var section = configuration.GetSection(LocalDbRegistration.ReplicationSectionPath); - - return !string.IsNullOrWhiteSpace(section["PeerAddress"]) - || !string.IsNullOrWhiteSpace(section["ApiKey"]); + // Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture + // changes and the nodes that stand down for a peer can never disagree. + return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration); } /// Opts every replicated table into change capture, baselining what is already there. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs index a0fc339d..80369a23 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs @@ -16,7 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines; /// /// Single-tag mode: cross-tag ctx.SetVirtualTag writes are dropped (logged) because /// fan-out between actors is owned by DependencyMuxActor, not by the eval engine. -/// Cycle detection + cascade ordering live in ; this adapter +/// Cycle detection lives in the deploy gate (DependencyGraph); this adapter /// stays single-tag scoped to keep 's message loop simple. /// /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs index eee64688..7693ac45 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs @@ -579,7 +579,14 @@ public sealed class AddressSpaceApplier } /// - /// Auto-provision the historian for the added historized equipment tags. Runs on the OPC UA + /// Resolve the historian name EXACTLY as MaterialiseRawSubtree does: a null/blank override falls + /// back to the RawPath (== NodeId). + /// + private static string ResolveHistorianName(RawTagPlan tag) => + string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.NodeId : tag.HistorianTagname; + + /// + /// Auto-provision the historian for the added (and newly-historized / renamed-historian) raw tags. Runs on the OPC UA /// publish actor's pinned thread, so the synchronous portion is kept to building the request /// list only and the gateway round-trip is dispatched fire-and-forget. The whole hook is wrapped /// in try/catch — a synchronously-throwing provisioner (or any request-building fault) is @@ -597,22 +604,39 @@ public sealed class AddressSpaceApplier foreach (var tag in plan.AddedRawTags) { if (!tag.IsHistorized || tag.Alarm is not null) continue; + AddRequest(tag); + } - // Parse the driver-agnostic data type from the tag's DataType string. An unparseable - // type is skipped (logged at Debug) rather than faulting the hook. + // #600.1: a historization flip (false->true) or a historian-name / data-type change on an + // EXISTING raw tag never appears in AddedRawTags, so EnsureTags never fired for it. EnsureTags is + // an upsert, so re-ensuring is safe; an unchanged pair is a no-op. Only raw deltas are used - a + // UNS mirror delta must not double-provision the same historian tag. + foreach (var d in plan.ChangedRawTags) + { + var cur = d.Current; + if (!cur.IsHistorized || cur.Alarm is not null) continue; + var prev = d.Previous; + var wasProvisioned = prev.IsHistorized && prev.Alarm is null; + if (wasProvisioned + && ResolveHistorianName(prev) == ResolveHistorianName(cur) + && string.Equals(prev.DataType, cur.DataType, StringComparison.OrdinalIgnoreCase)) + continue; + AddRequest(cur); + } + + void AddRequest(RawTagPlan tag) + { + // An unparseable data type is skipped (logged at Debug) rather than faulting the hook. if (!Enum.TryParse(tag.DataType, ignoreCase: true, out var dataType)) { _logger.LogDebug( - "AddressSpaceApplier: skipping historian provisioning for an added historized tag whose data type '{DataType}' is not a DriverDataType", + "AddressSpaceApplier: skipping historian provisioning for a historized tag whose data type '{DataType}' is not a DriverDataType", tag.DataType); - continue; + return; } - // Resolve the historian name EXACTLY as MaterialiseRawSubtree does: a null/blank override - // falls back to the RawPath (== NodeId). - var historianName = string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.NodeId : tag.HistorianTagname; (requests ??= new List()).Add( - new HistorianTagProvisionRequest(historianName, dataType, EngineeringUnit: null, Description: tag.Name)); + new HistorianTagProvisionRequest(ResolveHistorianName(tag), dataType, EngineeringUnit: null, Description: tag.Name)); } if (requests is null) return; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs new file mode 100644 index 00000000..e4106f55 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs @@ -0,0 +1,43 @@ +using Microsoft.Extensions.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Runtime; + +/// +/// The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree +/// on it, and used to carry separate predicates that diverged: the Host's table-registration step +/// (LocalDbSetup.ReplicationIsConfigured - which nodes capture changes) and the alarm-historian +/// drain gate (AddAlarmHistorian - which nodes stand down in favour of a peer). Lives in Runtime +/// because Host references Runtime and not the reverse. +/// +/// +/// +/// Predicate: LocalDb:Replication:PeerAddress OR LocalDb:Replication:ApiKey is +/// non-blank. The OR is required: only the dialling half of a pair sets PeerAddress, while +/// the passive half carries the shared ApiKey (plus a listen port) and nothing else. +/// +/// +/// Why the drain gate takes the same predicate (it previously used PeerAddress OR +/// SyncListenPort). The gate's question is "does a peer hold the same rows?". The passive +/// sync endpoint is fail-closed on the key, so a node with a listen port but no ApiKey rejects +/// every inbound stream and shares its queue with nobody - gating it would park its alarm events +/// behind a Primary that cannot deliver them. Conversely an ApiKey-only passive node IS +/// registered and does receive its peer's rows, so it must be gated or both halves drain. The port +/// is not evidence of a peer; the key is. +/// +/// +public static class LocalDbReplicationPolicy +{ + /// Configuration section holding the LocalDb replication options. + public const string ReplicationSectionPath = "LocalDb:Replication"; + + /// Whether this node's LocalDb replicates to a pair peer. + /// The application configuration. + /// true when PeerAddress or ApiKey is set under the replication section. + public static bool IsConfigured(IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + var section = configuration.GetSection(ReplicationSectionPath); + return !string.IsNullOrWhiteSpace(section["PeerAddress"]) + || !string.IsNullOrWhiteSpace(section["ApiKey"]); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index ee611e63..a530ce62 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -177,12 +177,11 @@ public static class ServiceCollectionExtensions // possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what // keeps the two scopes from disagreeing. // BOTH halves of a pair share the queue, but only one of them dials: the initiator sets - // Replication:PeerAddress, its partner only sets SyncListenPort and waits. Testing the dial - // side alone would leave the listening half permanently ungated — one drainer per pair by - // accident rather than by role, and the wrong one whenever the roles swap. - var replicated = - !string.IsNullOrWhiteSpace(configuration["LocalDb:Replication:PeerAddress"]) - || !string.IsNullOrWhiteSpace(configuration["LocalDb:SyncListenPort"]); + // Replication:PeerAddress, its partner carries the shared ApiKey and waits. The predicate is + // LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture + // changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is + // fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining. + var replicated = LocalDbReplicationPolicy.IsConfigured(configuration); if (!replicated) { 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 c2c7a8fd..c050bf8e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs @@ -366,7 +366,7 @@ public sealed class VirtualTagActor : ReceiveActor _virtualTagId, _hasLastValue ? _lastValue : null, timestampUtc, CorrelationId.NewId(), OpcUaStatus.BadInternalError)); } - /// Inputs-ready gate (mirrors the dormant VirtualTagEngine.AreInputsReady): true once + /// Inputs-ready gate (the retired cascade engine's rule): true once /// every declared dependency ref has been seen at least once. Called once per failure — the ref list /// is small, so no caching. A vacuously-true empty ref list means a no-dependency script degrades on /// its first failure. diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptSandboxBuildTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptSandboxBuildTests.cs index a3d3e212..5231af8b 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptSandboxBuildTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptSandboxBuildTests.cs @@ -57,4 +57,18 @@ public sealed class ScriptSandboxBuildTests """return (double)ctx.GetTag("X").Value;"""); evaluator.ShouldNotBeNull(); } + + /// #603: the BCL reference set is built once per process, so two Build calls share the same + /// MetadataReference instances rather than re-reading ~170 DLLs per compile. + [Fact] + public void Build_reuses_metadata_reference_instances_across_calls() + { + var first = ScriptSandbox.Build(typeof(ScriptContext)); + var second = ScriptSandbox.Build(typeof(ScriptContext)); + + first.References.Count.ShouldBeGreaterThan(50); + first.References.Count.ShouldBe(second.References.Count); + for (var i = 0; i < first.References.Count; i++) + ReferenceEquals(first.References[i], second.References[i]).ShouldBeTrue($"reference {i} was rebuilt"); + } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/FakeUpstream.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/FakeUpstream.cs deleted file mode 100644 index b19aea0d..00000000 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/FakeUpstream.cs +++ /dev/null @@ -1,91 +0,0 @@ -using System.Collections.Concurrent; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests; - -/// -/// In-memory for tests. Seed tag values via -/// , push changes via . Tracks subscriptions so -/// tests can assert the engine disposes them on reload / shutdown. -/// -public sealed class FakeUpstream : ITagUpstreamSource -{ - private readonly ConcurrentDictionary _values = new(StringComparer.Ordinal); - private readonly ConcurrentDictionary>> _subs = new(StringComparer.Ordinal); - - /// Gets the number of active subscriptions. - public int ActiveSubscriptionCount { get; private set; } - - /// Sets a tag value without notifying subscribers. - /// The path to the tag. - /// The value to set. - /// The OPC UA status code for the value. - public void Set(string path, object value, uint statusCode = 0u) - { - var now = DateTime.UtcNow; - _values[path] = new DataValueSnapshot(value, statusCode, now, now); - } - - /// Sets a tag value and notifies all subscribers. - /// The path to the tag. - /// The value to set. - /// The OPC UA status code for the value. - public void Push(string path, object value, uint statusCode = 0u) - { - Set(path, value, statusCode); - if (_subs.TryGetValue(path, out var list)) - { - Action[] snap; - lock (list) { snap = list.ToArray(); } - foreach (var obs in snap) obs(path, _values[path]); - } - } - - /// Reads the current value of a tag. - /// The path to the tag. - public DataValueSnapshot ReadTag(string path) - => _values.TryGetValue(path, out var v) - ? v - : new DataValueSnapshot(null, 0x80340000u, null, DateTime.UtcNow); - - /// Subscribes to tag value changes. - /// The path to the tag. - /// The callback to invoke when the tag value changes. - public IDisposable SubscribeTag(string path, Action observer) - { - var list = _subs.GetOrAdd(path, _ => []); - lock (list) { list.Add(observer); } - ActiveSubscriptionCount++; - return new Unsub(this, path, observer); - } - - private sealed class Unsub : IDisposable - { - private readonly FakeUpstream _up; - private readonly string _path; - private readonly Action _observer; - - /// Initializes a new subscription unsubscription handler. - /// The parent FakeUpstream instance. - /// The tag path. - /// The observer callback to unsubscribe. - public Unsub(FakeUpstream up, string path, Action observer) - { - _up = up; _path = path; _observer = observer; - } - - /// Unsubscribes from tag value changes. - public void Dispose() - { - if (_up._subs.TryGetValue(_path, out var list)) - { - lock (list) - { - if (list.Remove(_observer)) - _up.ActiveSubscriptionCount--; - } - } - } - } -} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/TimerTriggerSchedulerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/TimerTriggerSchedulerTests.cs deleted file mode 100644 index f21b64d2..00000000 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/TimerTriggerSchedulerTests.cs +++ /dev/null @@ -1,203 +0,0 @@ -using Serilog; -using Shouldly; -using Xunit; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Core.Scripting; -using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests; - -[Trait("Category", "Unit")] -public sealed class TimerTriggerSchedulerTests -{ - /// Verifies that timer interval causes periodic reevaluation of virtual tags. - [Fact] - public async Task Timer_interval_causes_periodic_reevaluation() - { - var up = new FakeUpstream(); - // Counter source — re-eval should pick up new value each tick. - var counter = 0; - var logger = new LoggerConfiguration().CreateLogger(); - - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), - logger); - - engine.Load([new VirtualTagDefinition( - "Counter", DriverDataType.Int32, - """return ctx.Now.Millisecond;""", // changes on every evaluation - ChangeTriggered: false, - TimerInterval: TimeSpan.FromMilliseconds(100))]); - - using var sched = new TimerTriggerScheduler(engine, logger); - sched.Start([new VirtualTagDefinition( - "Counter", DriverDataType.Int32, - """return ctx.Now.Millisecond;""", - ChangeTriggered: false, - TimerInterval: TimeSpan.FromMilliseconds(100))]); - - // Watch the value change across ticks. - var snapshots = new List(); - using var sub = engine.Subscribe("Counter", (_, v) => snapshots.Add(v.Value)); - - await Task.Delay(500); - - snapshots.Count.ShouldBeGreaterThanOrEqualTo(3, "At least 3 ticks in 500ms at 100ms cadence"); - } - - /// Verifies that tags without TimerInterval are not scheduled. - [Fact] - public async Task Tags_without_TimerInterval_not_scheduled() - { - var up = new FakeUpstream(); - var logger = new LoggerConfiguration().CreateLogger(); - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), logger); - engine.Load([new VirtualTagDefinition( - "NoTimer", DriverDataType.Int32, """return 1;""")]); - - using var sched = new TimerTriggerScheduler(engine, logger); - sched.Start([new VirtualTagDefinition( - "NoTimer", DriverDataType.Int32, """return 1;""")]); - - var events = new List(); - using var sub = engine.Subscribe("NoTimer", (_, v) => events.Add((int)(v.Value ?? 0))); - - await Task.Delay(300); - events.Count.ShouldBe(0, "No TimerInterval = no timer ticks"); - } - - /// Verifies that Start groups tags by interval into shared timers. - [Fact] - public void Start_groups_tags_by_interval_into_shared_timers() - { - // Smoke test — Start on a definition list with two distinct intervals must not - // throw. Group count matches unique intervals. - var up = new FakeUpstream(); - var logger = new LoggerConfiguration().CreateLogger(); - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), logger); - engine.Load([ - new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromSeconds(1)), - new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""", - TimerInterval: TimeSpan.FromSeconds(5)), - new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""", - TimerInterval: TimeSpan.FromSeconds(1)), - ]); - - using var sched = new TimerTriggerScheduler(engine, logger); - Should.NotThrow(() => sched.Start(new[] - { - new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""", TimerInterval: TimeSpan.FromSeconds(1)), - new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""", TimerInterval: TimeSpan.FromSeconds(5)), - new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""", TimerInterval: TimeSpan.FromSeconds(1)), - })); - } - - // ----- Core.VirtualTags-007: timer ticks must not block pool threads and must skip when prior tick is still running ----- - - /// Verifies that tick is skipped when the prior tick for the same group is still running. - [Fact] - public async Task Tick_skips_when_prior_tick_for_the_same_group_is_still_running() - { - // Regression for Core.VirtualTags-007: if a single tick takes longer than the - // interval, subsequent timer callbacks must NOT each pin a thread-pool thread - // waiting on the same evaluation gate. The scheduler tracks an in-flight flag - // per group and skips a new tick when the prior one is still running. - var up = new FakeUpstream(); - up.Set("In", 1); - var logger = new LoggerConfiguration().CreateLogger(); - - // Slow script — each evaluation takes longer than several timer intervals. - const int slowMs = 250; - const int intervalMs = 50; - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), logger); - engine.Load([new VirtualTagDefinition( - "Slow", DriverDataType.Int32, - $$""" - var end = DateTime.UtcNow.AddMilliseconds({{slowMs}}); - while (DateTime.UtcNow < end) { } - return (int)ctx.GetTag("In").Value; - """, - ChangeTriggered: false, - TimerInterval: TimeSpan.FromMilliseconds(intervalMs))]); - - using var sched = new TimerTriggerScheduler(engine, logger); - sched.Start([new VirtualTagDefinition( - "Slow", DriverDataType.Int32, - "", - ChangeTriggered: false, - TimerInterval: TimeSpan.FromMilliseconds(intervalMs))]); - - // Wait long enough for many timer ticks at 50ms while one evaluation - // (~250ms each) holds the engine. Window is 600ms ~ 12 ticks. - await Task.Delay(600); - - // With the fix in place, ticks that fire while the previous one for the same - // group is still running are skipped. The skipped count must be measurable; if - // SkippedTickCount is still 0 after 600ms with ~12 ticks fired and a 250ms eval, - // the fix is not working — at minimum 3-4 ticks must have been skipped. - sched.SkippedTickCount.ShouldBeGreaterThan(2, - "ticks that fire while the prior tick for the same group is still running must be skipped"); - } - - /// Verifies that a disposed scheduler stops firing timers. - [Fact] - public void Disposed_scheduler_stops_firing() - { - var up = new FakeUpstream(); - var logger = new LoggerConfiguration().CreateLogger(); - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), logger); - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromMilliseconds(50))]); - - var sched = new TimerTriggerScheduler(engine, logger); - sched.Start([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromMilliseconds(50))]); - sched.Dispose(); - - // After dispose, second Start throws ObjectDisposedException. - Should.Throw(() => - sched.Start([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromMilliseconds(50))])); - } - - // ----- Core.VirtualTags-015: Start must not be callable twice on the same instance ----- - - /// - /// Verifies that a second call to Start on the same (non-disposed) scheduler - /// throws InvalidOperationException. Without this guard, calling Start twice - /// appends a duplicate set of timers and TickGroups, causing every timer-group - /// to fire and evaluate twice per interval with no diagnostic. - /// - [Fact] - public void Start_called_twice_throws_InvalidOperationException() - { - // Regression for Core.VirtualTags-015: without the _started guard, two Start - // calls silently double the timer count and the evaluation frequency. - var up = new FakeUpstream(); - var logger = new LoggerConfiguration().CreateLogger(); - using var engine = new VirtualTagEngine(up, - new ScriptLoggerFactory(logger), logger); - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromSeconds(10))]); - - using var sched = new TimerTriggerScheduler(engine, logger); - sched.Start([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromSeconds(10))]); - - // Second call on the same live instance must throw — not silently create duplicates. - Should.Throw(() => - sched.Start([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""", - TimerInterval: TimeSpan.FromSeconds(10))])); - } -} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagEngineTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagEngineTests.cs deleted file mode 100644 index 4d3364f3..00000000 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagEngineTests.cs +++ /dev/null @@ -1,800 +0,0 @@ -using Serilog; -using Shouldly; -using Xunit; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Core.Scripting; -using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests; - -/// -/// End-to-end VirtualTagEngine behavior: load config, subscribe to upstream, -/// evaluate on change, cascade through dependent virtual tags, timer-driven -/// re-evaluation, error isolation, historize flag, cycle rejection. -/// -[Trait("Category", "Unit")] -public sealed class VirtualTagEngineTests -{ - private static VirtualTagEngine Build( - FakeUpstream upstream, - IHistoryWriter? history = null, - TimeSpan? scriptTimeout = null, - Func? clock = null) - { - var rootLogger = new LoggerConfiguration().CreateLogger(); - return new VirtualTagEngine( - upstream, - new ScriptLoggerFactory(rootLogger), - rootLogger, - history, - clock, - scriptTimeout); - } - - /// Verifies that a simple virtual tag script can read an upstream tag and return a coerced value. - [Fact] - public async Task Simple_script_reads_upstream_and_returns_coerced_value() - { - var up = new FakeUpstream(); - up.Set("InTag", 10.0); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - Path: "LineRate", - DataType: DriverDataType.Float64, - ScriptSource: """return (double)ctx.GetTag("InTag").Value * 2.0;""")]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var result = engine.Read("LineRate"); - result.StatusCode.ShouldBe(0u); - result.Value.ShouldBe(20.0); - } - - /// Verifies that an upstream tag change triggers a cascade re-evaluation through two levels of dependent virtual tags. - [Fact] - public async Task Upstream_change_triggers_cascade_through_two_levels() - { - var up = new FakeUpstream(); - up.Set("A", 1.0); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("B", DriverDataType.Float64, - """return (double)ctx.GetTag("A").Value + 10.0;"""), - new VirtualTagDefinition("C", DriverDataType.Float64, - """return (double)ctx.GetTag("B").Value * 2.0;"""), - ]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("B").Value.ShouldBe(11.0); - engine.Read("C").Value.ShouldBe(22.0); - - // Change upstream — cascade should recompute B (11→15.0) then C (30.0). - // Both B and C are updated in the same CascadeAsync call (topological order: - // B then C), but we must wait for each independently: the WaitForConditionAsync - // on B returns as soon as _valueCache["B"] is set (before the semaphore is - // released for C's evaluation), so asserting C immediately after the B-wait - // races against C's still-in-progress evaluation. Wait for C explicitly. - up.Push("A", 5.0); - await WaitForConditionAsync(() => Equals(engine.Read("B").Value, 15.0)); - await WaitForConditionAsync(() => Equals(engine.Read("C").Value, 30.0)); - engine.Read("B").Value.ShouldBe(15.0); - engine.Read("C").Value.ShouldBe(30.0); - } - - /// Verifies that a circular dependency among virtual tags is rejected at load time. - [Fact] - public async Task Cycle_in_virtual_tags_rejected_at_Load() - { - var up = new FakeUpstream(); - using var engine = Build(up); - - Should.Throw(() => engine.Load([ - new VirtualTagDefinition("A", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value + 1;"""), - new VirtualTagDefinition("B", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value + 1;"""), - ])); - await Task.CompletedTask; - } - - /// Verifies that script compilation errors surface at load time with all failures aggregated. - [Fact] - public async Task Script_compile_error_surfaces_at_Load_with_all_failures() - { - var up = new FakeUpstream(); - using var engine = Build(up); - - var ex = Should.Throw(() => engine.Load([ - new VirtualTagDefinition("A", DriverDataType.Int32, """return undefinedIdentifier;"""), - new VirtualTagDefinition("B", DriverDataType.Int32, """return 42;"""), - new VirtualTagDefinition("C", DriverDataType.Int32, """var x = anotherUndefined; return x;"""), - ])); - ex.Message.ShouldContain("2 script(s) did not compile"); - ex.Message.ShouldContain("A"); - ex.Message.ShouldContain("C"); - await Task.CompletedTask; - } - - /// Verifies that a runtime exception in one virtual tag's script is isolated and does not affect other tags. - [Fact] - public async Task Script_runtime_exception_isolates_to_owning_tag() - { - var up = new FakeUpstream(); - up.Set("OK", 10); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("GoodTag", DriverDataType.Int32, - """return (int)ctx.GetTag("OK").Value * 2;"""), - new VirtualTagDefinition("BadTag", DriverDataType.Int32, - """throw new InvalidOperationException("boom");"""), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - engine.Read("GoodTag").StatusCode.ShouldBe(0u); - engine.Read("GoodTag").Value.ShouldBe(20); - engine.Read("BadTag").StatusCode.ShouldBe(0x80020000u, "BadInternalError for thrown script"); - engine.Read("BadTag").Value.ShouldBeNull(); - } - - /// Verifies that script timeout is mapped to BadInternalError status without killing the engine. - [Fact] - public async Task Timeout_maps_to_BadInternalError_without_killing_the_engine() - { - var up = new FakeUpstream(); - using var engine = Build(up, scriptTimeout: TimeSpan.FromMilliseconds(30)); - - engine.Load([ - new VirtualTagDefinition("Hang", DriverDataType.Int32, """ - var end = DateTime.UtcNow.AddSeconds(5); - while (DateTime.UtcNow < end) { } - return 1; - """), - new VirtualTagDefinition("Ok", DriverDataType.Int32, """return 42;"""), - ]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("Hang").StatusCode.ShouldBe(0x80020000u); - engine.Read("Ok").Value.ShouldBe(42); - } - - /// Verifies that subscribers receive notifications when the engine emits value changes. - [Fact] - public async Task Subscribers_receive_engine_emitted_changes() - { - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value + 100;""")]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var received = new List(); - using var sub = engine.Subscribe("Out", (p, v) => received.Add(v)); - - up.Push("In", 5); - await WaitForConditionAsync(() => received.Count >= 1); - - received[^1].Value.ShouldBe(105); - } - - /// Verifies that the historize flag routes virtual tag values to the history writer. - [Fact] - public async Task Historize_flag_routes_to_history_writer() - { - var recorded = new List<(string, DataValueSnapshot)>(); - var history = new TestHistory(recorded); - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up, history); - - engine.Load([ - new VirtualTagDefinition("H", DriverDataType.Int32, - """return (int)ctx.GetTag("In").Value + 1;""", Historize: true), - new VirtualTagDefinition("NoH", DriverDataType.Int32, - """return (int)ctx.GetTag("In").Value - 1;""", Historize: false), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - recorded.Select(p => p.Item1).ShouldContain("H"); - recorded.Select(p => p.Item1).ShouldNotContain("NoH"); - } - - /// Verifies that upstream pushes are ignored when change-driven is false. - [Fact] - public async Task Change_driven_false_ignores_upstream_push() - { - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up); - engine.Load([new VirtualTagDefinition( - "Manual", DriverDataType.Int32, - """return (int)ctx.GetTag("In").Value * 10;""", - ChangeTriggered: false)]); - - // Initial eval seeds the value. - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("Manual").Value.ShouldBe(10); - - // Upstream change fires but change-driven is off — no recompute. - up.Push("In", 99); - await Task.Delay(100); - engine.Read("Manual").Value.ShouldBe(10, "change-driven=false ignores upstream deltas"); - } - - /// Verifies that reloading the engine replaces existing tags and resubscribes to upstream sources cleanly. - [Fact] - public async Task Reload_replaces_existing_tags_and_resubscribes_cleanly() - { - var up = new FakeUpstream(); - up.Set("A", 1); - up.Set("B", 2); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value * 2;""")]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("T").Value.ShouldBe(2); - up.ActiveSubscriptionCount.ShouldBe(1); - - // Reload — T now depends on B instead of A. - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value * 3;""")]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("T").Value.ShouldBe(6); - up.ActiveSubscriptionCount.ShouldBe(1, "previous subscription on A must be disposed"); - await Task.CompletedTask; - } - - /// Verifies that disposing the engine releases all upstream subscriptions. - [Fact] - public async Task Dispose_releases_upstream_subscriptions() - { - var up = new FakeUpstream(); - up.Set("A", 1); - var engine = Build(up); - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value;""")]); - up.ActiveSubscriptionCount.ShouldBe(1); - - engine.Dispose(); - up.ActiveSubscriptionCount.ShouldBe(0); - await Task.CompletedTask; - } - - /// Verifies that SetVirtualTag called within a script updates the target and triggers observers. - [Fact] - public async Task SetVirtualTag_within_script_updates_target_and_triggers_observers() - { - var up = new FakeUpstream(); - up.Set("In", 5); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("Target", DriverDataType.Int32, - """return 0;""", ChangeTriggered: false), // placeholder value, operator-written via SetVirtualTag - new VirtualTagDefinition("Driver", DriverDataType.Int32, - """ - var v = (int)ctx.GetTag("In").Value; - ctx.SetVirtualTag("Target", v * 100); - return v; - """), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - engine.Read("Target").Value.ShouldBe(500); - engine.Read("Driver").Value.ShouldBe(5); - } - - /// Verifies that SetVirtualTag within a script cascades to change-triggered dependents of the written tag. - [Fact] - public async Task SetVirtualTag_within_script_cascades_to_dependents_of_the_written_tag() - { - // Regression for Core.VirtualTags-001: a ctx.SetVirtualTag write must schedule - // the change-trigger cascade for tags depending on the written path, exactly - // like an upstream delta. "Dependent" reads "Target" and is ChangeTriggered, so - // a write to Target via ctx.SetVirtualTag must re-evaluate Dependent. - var up = new FakeUpstream(); - up.Set("In", 5); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("Target", DriverDataType.Int32, - """return 0;""", ChangeTriggered: false), // operator-written via SetVirtualTag - new VirtualTagDefinition("Dependent", DriverDataType.Int32, - """return (int)ctx.GetTag("Target").Value + 1;""", ChangeTriggered: true), - new VirtualTagDefinition("Writer", DriverDataType.Int32, - """ - var v = (int)ctx.GetTag("In").Value; - ctx.SetVirtualTag("Target", v * 100); - return v; - """), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - // Writer set Target = 500; the cascade must have re-evaluated Dependent to 501. - engine.Read("Target").Value.ShouldBe(500); - await WaitForConditionAsync(() => Equals(engine.Read("Dependent").Value, 501)); - engine.Read("Dependent").Value.ShouldBe(501, - "a ctx.SetVirtualTag write must cascade to change-triggered dependents"); - } - - /// Verifies that type coercion from script double to configured int32 works correctly. - [Fact] - public async Task Type_coercion_from_script_double_to_config_int32() - { - var up = new FakeUpstream(); - up.Set("In", 3.7); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "Rounded", DriverDataType.Int32, - """return (double)ctx.GetTag("In").Value;""")]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - engine.Read("Rounded").Value.ShouldBe(4, "Convert.ToInt32 rounds 3.7 to 4"); - } - - // ----- Core.VirtualTags-012: previously-missing coverage ----- - - /// Verifies that the AreInputsReady guard publishes BadWaitingForInitialData when upstream tags have bad status. - [Fact] - public async Task AreInputsReady_guard_publishes_BadWaitingForInitialData_when_upstream_is_bad() - { - // Arrange: upstream tag is Bad-quality (not yet available). - var up = new FakeUpstream(); - up.Set("BadIn", null!, 0x80000000u); // bad status, null value - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "Derived", DriverDataType.Int32, - """return (int)ctx.GetTag("BadIn").Value * 2;""")]); - - // Act: evaluate — inputs are not ready. - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - // Assert: tag publishes BadWaitingForInitialData, not a stale null/Good. - var result = engine.Read("Derived"); - result.StatusCode.ShouldBe(0x80320000u, "BadWaitingForInitialData expected when inputs are bad"); - result.Value.ShouldBeNull(); - } - - /// Verifies that the AreInputsReady guard recovers when an upstream tag transitions from bad to good status. - [Fact] - public async Task AreInputsReady_guard_publishes_BadWaitingForInitialData_then_recovers_when_upstream_becomes_good() - { - // Arrange: upstream tag starts absent (null/Bad). - var up = new FakeUpstream(); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "Derived", DriverDataType.Int32, - """return (int)ctx.GetTag("In").Value + 1;""")]); - - // First evaluation: upstream not ready → BadWaitingForInitialData. - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("Derived").StatusCode.ShouldBe(0x80320000u); - - // Upstream becomes available. - up.Push("In", 10); - await WaitForConditionAsync(() => engine.Read("Derived").StatusCode == 0u); - - // Tag should now have a Good value. - engine.Read("Derived").StatusCode.ShouldBe(0u); - engine.Read("Derived").Value.ShouldBe(11); - } - - /// Verifies that SetVirtualTag cascades to change-triggered dependents. - [Fact] - public async Task SetVirtualTag_cascades_to_change_triggered_dependent() - { - // Arrange: "Writer" writes to "Target"; "Consumer" reads "Target" and is change-triggered. - var up = new FakeUpstream(); - up.Set("In", 3); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("Target", DriverDataType.Int32, - """return 0;""", ChangeTriggered: false), - new VirtualTagDefinition("Writer", DriverDataType.Int32, - """ - var v = (int)ctx.GetTag("In").Value; - ctx.SetVirtualTag("Target", v * 10); - return v; - """), - new VirtualTagDefinition("Consumer", DriverDataType.Int32, - """return (int)ctx.GetTag("Target").Value + 1;""", - ChangeTriggered: true), - ]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - // Writer sets Target = 30; Consumer should cascade and compute 31. - await WaitForConditionAsync(() => engine.Read("Consumer").Value is int v && v == 31); - engine.Read("Target").Value.ShouldBe(30); - engine.Read("Consumer").Value.ShouldBe(31); - } - - /// Verifies that SetVirtualTag calls targeting unregistered paths are caught at load time. - [Fact] - public async Task SetVirtualTag_on_non_registered_path_is_caught_at_Load() - { - // Originally validated the runtime warning-and-drop branch in OnScriptSetVirtualTag. - // After Core.VirtualTags-011 the static DependencyExtractor.Writes set is validated - // at Load time, so a literal-string write to a non-existent path is now rejected - // at publish — the dynamic warning path is reserved as a defensive guard for cases - // the static extractor cannot see (currently none, since dynamic paths are also - // rejected at extraction). - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up); - - Should.Throw(() => engine.Load([ - new VirtualTagDefinition("Writer", DriverDataType.Int32, - """ - ctx.SetVirtualTag("NonExistentPath", 99); - return (int)ctx.GetTag("In").Value; - """) - ])).Message.ShouldContain("NonExistentPath"); - await Task.CompletedTask; - } - - /// Verifies that EvaluateOneAsync throws ArgumentException when called for an unregistered path. - [Fact] - public async Task EvaluateOneAsync_throws_ArgumentException_for_unregistered_path() - { - var up = new FakeUpstream(); - using var engine = Build(up); - engine.Load([new VirtualTagDefinition("A", DriverDataType.Int32, """return 1;""")]); - - await Should.ThrowAsync(async () => - await engine.EvaluateOneAsync("NoSuchTag", TestContext.Current.CancellationToken)); - } - - /// Verifies that a type coercion failure maps to BadInternalError status. - [Fact] - public async Task CoerceResult_failure_maps_to_BadInternalError() - { - // Arrange: script returns an object that cannot be coerced to the declared type. - var up = new FakeUpstream(); - using var engine = Build(up); - - engine.Load([new VirtualTagDefinition( - "Bad", DriverDataType.Int32, - // Return a non-numeric string — Convert.ToInt32("not-a-number") throws. - """return "not-a-number";""")]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - // CoerceResult returns null on failure; the null propagates as BadInternalError. - engine.Read("Bad").StatusCode.ShouldBe(0x80020000u, "type-coercion failure must map to BadInternalError"); - engine.Read("Bad").Value.ShouldBeNull(); - } - - // ----- Core.VirtualTags-011: Writes target validation at Load time ----- - - /// Verifies that Load rejects scripts that write to unregistered virtual tag paths. - [Fact] - public async Task Load_rejects_script_writing_to_unregistered_virtual_tag_path() - { - // Regression for Core.VirtualTags-011: a script that calls - // ctx.SetVirtualTag("Typo", ...) must be caught at publish/load time rather than - // silently dropped at runtime, so operator typos surface as a publish failure. - var up = new FakeUpstream(); - using var engine = Build(up); - - var ex = Should.Throw(() => engine.Load([ - new VirtualTagDefinition("Writer", DriverDataType.Int32, - """ - ctx.SetVirtualTag("NonRegisteredTarget", 1); - return 0; - """), - new VirtualTagDefinition("RegisteredTarget", DriverDataType.Int32, - """return 1;"""), - ])); - ex.Message.ShouldContain("Writer"); - ex.Message.ShouldContain("NonRegisteredTarget"); - await Task.CompletedTask; - } - - /// Verifies that Load accepts scripts that write to registered virtual tag paths. - [Fact] - public async Task Load_accepts_script_writing_to_registered_virtual_tag_path() - { - // Companion to the rejection test: a write to a registered tag must continue to - // load successfully. - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up); - - // No throw — Writer writes to Target which is registered. - engine.Load([ - new VirtualTagDefinition("Target", DriverDataType.Int32, - """return 0;""", ChangeTriggered: false), - new VirtualTagDefinition("Writer", DriverDataType.Int32, - """ - ctx.SetVirtualTag("Target", (int)ctx.GetTag("In").Value); - return 0; - """), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - engine.Read("Target").Value.ShouldBe(1); - } - - // ----- Core.VirtualTags-006: empty observer list left in _observers map ----- - - /// Verifies that disposing the last subscriber for a path removes the empty observer list entry. - [Fact] - public void Subscribe_then_Unsub_prunes_empty_observer_list_for_path() - { - // Regression for Core.VirtualTags-006: disposing the last subscriber for a path - // must remove the dictionary entry so a long-running server with churning OPC UA - // subscriptions does not accumulate an unbounded number of empty List entries. - var up = new FakeUpstream(); - using var engine = Build(up); - engine.Load([new VirtualTagDefinition( - "T", DriverDataType.Int32, """return 1;""")]); - - // Subscribe, then immediately Dispose — both the only observer. - var sub1 = engine.Subscribe("T", (_, _) => { }); - var sub2 = engine.Subscribe("T", (_, _) => { }); - sub1.Dispose(); - sub2.Dispose(); - - // The internal map should no longer hold an entry for the path. - // Use the same ConcurrentDictionary type the engine uses; we check via reflection - // on the test-private field so this is robust to future renames inside engine. - var observersField = typeof(VirtualTagEngine).GetField( - "_observers", - System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance); - observersField.ShouldNotBeNull(); - var observers = observersField!.GetValue(engine); - observers.ShouldNotBeNull(); - var containsKey = observers!.GetType().GetMethod("ContainsKey")!; - var result = (bool)containsKey.Invoke(observers, new object[] { "T" })!; - result.ShouldBeFalse("disposing the last subscriber must remove the dictionary entry"); - } - - // ----- Core.VirtualTags-004: CoerceResult default arm leaks uncoerced values ----- - - /// Verifies that CoerceResult correctly handles Int16, UInt16, UInt32, and UInt64 types. - [Fact] - public async Task CoerceResult_handles_Int16_UInt16_UInt32_UInt64() - { - // Regression for Core.VirtualTags-004: before the fix, CoerceResult had a default - // arm that returned the script's raw double/string for these types, producing a - // type-mismatched DataValueSnapshot. Verify every integer DriverDataType the engine - // is allowed to declare coerces correctly. - var up = new FakeUpstream(); - using var engine = Build(up); - - engine.Load([ - new VirtualTagDefinition("AsInt16", DriverDataType.Int16, """return 7.0;"""), - new VirtualTagDefinition("AsUInt16", DriverDataType.UInt16, """return 8.0;"""), - new VirtualTagDefinition("AsUInt32", DriverDataType.UInt32, """return 9.0;"""), - new VirtualTagDefinition("AsUInt64", DriverDataType.UInt64, """return 10.0;"""), - ]); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - engine.Read("AsInt16").Value.ShouldBeOfType(); - engine.Read("AsInt16").Value.ShouldBe((short)7); - engine.Read("AsUInt16").Value.ShouldBeOfType(); - engine.Read("AsUInt16").Value.ShouldBe((ushort)8); - engine.Read("AsUInt32").Value.ShouldBeOfType(); - engine.Read("AsUInt32").Value.ShouldBe((uint)9); - engine.Read("AsUInt64").Value.ShouldBeOfType(); - engine.Read("AsUInt64").Value.ShouldBe((ulong)10); - } - - /// Verifies that Load rejects virtual tag definitions with unsupported DriverDataType values. - [Fact] - public async Task Load_rejects_definition_with_unsupported_DriverDataType() - { - // Regression for Core.VirtualTags-004: any DriverDataType that CoerceResult cannot - // honour must be rejected at Load time so an operator typo (or a future enum - // member added without coercion support) does not silently emit a type-mismatched - // value to OPC UA clients. Reference is unsupported for virtual tags (the engine - // does not synthesize Galaxy attribute references). - var up = new FakeUpstream(); - using var engine = Build(up); - - var ex = Should.Throw(() => engine.Load([ - new VirtualTagDefinition("Ref", DriverDataType.Reference, """return "Some.Attribute";"""), - ])); - ex.Message.ShouldContain("Reference"); - ex.Message.ShouldContain("Ref"); - await Task.CompletedTask; - } - - /// Verifies that Load rejects duplicate virtual tag paths with an aggregated error message. - [Fact] - public async Task Load_rejects_duplicate_path_with_aggregated_error() - { - var up = new FakeUpstream(); - using var engine = Build(up); - - var ex = Should.Throw(() => engine.Load([ - new VirtualTagDefinition("Dup", DriverDataType.Int32, """return 1;"""), - new VirtualTagDefinition("Dup", DriverDataType.Int32, """return 2;"""), - new VirtualTagDefinition("Good", DriverDataType.Int32, """return 3;"""), - ])); - ex.Message.ShouldContain("Dup"); - ex.Message.ShouldContain("duplicate"); - await Task.CompletedTask; - } - - /// Verifies that Read returns BadNodeIdUnknown for all paths before Load is called. - [Fact] - public void Read_before_Load_returns_BadNodeIdUnknown() - { - var up = new FakeUpstream(); - using var engine = Build(up); - - // Read is allowed before Load — it just returns BadNodeIdUnknown for everything. - var result = engine.Read("AnyPath"); - result.StatusCode.ShouldBe(0x80340000u, "BadNodeIdUnknown before Load"); - } - - /// Verifies that EvaluateOneAsync throws InvalidOperationException when called before Load. - [Fact] - public void EvaluateOneAsync_before_Load_throws_InvalidOperationException() - { - var up = new FakeUpstream(); - using var engine = Build(up); - - Should.Throw(() => - engine.EvaluateOneAsync("A").GetAwaiter().GetResult()); - } - - /// Verifies that Subscribe does not throw when called before Load. - [Fact] - public void Subscribe_before_Load_does_not_throw() - { - // Subscribe uses GetOrAdd and does not call EnsureLoaded — it should work - // (returns an Unsub handle) without a Load. The observer won't fire because - // no tag is registered, but it must not throw. - var up = new FakeUpstream(); - using var engine = Build(up); - - var fired = false; - var sub = engine.Subscribe("AnyPath", (_, _) => fired = true); - sub.ShouldNotBeNull(); - sub.Dispose(); - fired.ShouldBeFalse("no evaluation has happened, observer must not fire"); - } - - // ----- Core.VirtualTags-014: AreInputsReady must not block on Good-quality null value ----- - - /// - /// Verifies that a Good-quality upstream snapshot with a null Value (e.g. a - /// script returning null) does not permanently block downstream change-triggered - /// dependents at BadWaitingForInitialData. A Good-quality null means "the - /// upstream is online and returned null" — readiness is determined by StatusCode, - /// not by the value's nullability. - /// - [Fact] - public async Task AreInputsReady_Good_quality_null_upstream_does_not_block_downstream() - { - // Regression for Core.VirtualTags-014: before the fix, AreInputsReady checked - // kv.Value.Value is null, so a Good-quality DataValueSnapshot(null, 0u, ...) — - // produced by an upstream script returning null — kept every downstream stuck at - // BadWaitingForInitialData forever. The fix gates only on StatusCode (bit 31). - var up = new FakeUpstream(); - up.Set("In", 1); - using var engine = Build(up); - - engine.Load([ - // Producer script returns null — stores DataValueSnapshot(null, Good, ...). - new VirtualTagDefinition("NullProducer", DriverDataType.String, - """return null;"""), - // Consumer depends on NullProducer and is change-triggered. - // With the bug: Consumer is stuck at BadWaitingForInitialData. - // After the fix: Consumer evaluates and surfaces the NRE as BadInternalError - // (since it casts null unconditionally), which is the correct sentinel for - // "the script ran but produced an error" rather than "inputs not ready". - new VirtualTagDefinition("Consumer", DriverDataType.String, - """return (string)ctx.GetTag("NullProducer").Value + "_suffix";""", - ChangeTriggered: true), - ]); - - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - // NullProducer should be Good with null value (script returned null). - var producerResult = engine.Read("NullProducer"); - producerResult.StatusCode.ShouldBe(0u, "Good quality: script returned null cleanly"); - producerResult.Value.ShouldBeNull(); - - // Consumer must NOT be stuck at BadWaitingForInitialData — it should have - // evaluated (and produced BadInternalError from the cast of null + "_suffix"). - var consumerResult = engine.Read("Consumer"); - consumerResult.StatusCode.ShouldNotBe(0x80320000u, - "Consumer must not be stuck at BadWaitingForInitialData when its upstream is Good-quality"); - } - - private static async Task WaitForConditionAsync(Func cond, int timeoutMs = 2000) - { - var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); - while (DateTime.UtcNow < deadline) - { - if (cond()) return; - await Task.Delay(25); - } - throw new TimeoutException("Condition did not become true in time"); - } - - /// Test implementation of IHistoryWriter that records values to a buffer. - private sealed class TestHistory : IHistoryWriter - { - private readonly List<(string, DataValueSnapshot)> _buf; - - /// Initializes a new instance of the TestHistory class with the specified buffer. - /// The list to store recorded history entries. - public TestHistory(List<(string, DataValueSnapshot)> buf) => _buf = buf; - - /// Records a virtual tag path and value snapshot to the history buffer. - /// The virtual tag path. - /// The data value snapshot to record. - public void Record(string path, DataValueSnapshot value) - { - lock (_buf) { _buf.Add((path, value)); } - } - } - - // --- Core.Scripting-016: engine routes compiles through CompiledScriptCache --- - - /// Verifies that disposing the engine unloads the compiled script assembly. - [Fact] - public void Dispose_unloads_compiled_script_assembly() - { - // Pre-fix the engine called ScriptEvaluator.Compile directly, so the emitted - // script assembly's ALC stayed loaded for the process lifetime. After the fix - // the engine routes through CompiledScriptCache; engine Dispose triggers cache - // Dispose which unloads every cached evaluator's ALC. Assert via WeakReference - // + GC that the assembly is actually reclaimed. - var weak = CompileTagAndCaptureWeak(); - for (int i = 0; i < 10 && weak.IsAlive; i++) - { - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); - } - weak.IsAlive.ShouldBeFalse( - "engine Dispose must release the compiled-tag assembly via " + - "CompiledScriptCache (Core.Scripting-016). If this fails the engine is " + - "back to calling ScriptEvaluator.Compile directly and -008's headline " + - "fix doesn't run in production."); - } - - [System.Runtime.CompilerServices.MethodImpl( - System.Runtime.CompilerServices.MethodImplOptions.NoInlining)] - private static WeakReference CompileTagAndCaptureWeak() - { - var up = new FakeUpstream(); - up.Set("X", 10); - var engine = Build(up); - engine.Load([new VirtualTagDefinition( - Path: "Doubled", - DataType: DriverDataType.Float64, - ScriptSource: """return (double)ctx.GetTag("X").Value * 2.0;""")]); - - // Reach into the engine's compile cache via reflection — internals scoped to - // Tests-via-InternalsVisibleTo and the field is private. - var cacheField = typeof(VirtualTagEngine).GetField( - "_compileCache", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!; - var cache = cacheField.GetValue(engine)!; - var cacheDictField = cache.GetType().GetField( - "_cache", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!; - var cacheDict = (System.Collections.IDictionary)cacheDictField.GetValue(cache)!; - var lazy = cacheDict.Values.Cast().Single(); - var lazyValueProp = lazy.GetType().GetProperty("Value")!; - var evaluator = lazyValueProp.GetValue(lazy)!; - var funcField = evaluator.GetType().GetField( - "_func", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!; - var del = (Delegate)funcField.GetValue(evaluator)!; - var weak = new WeakReference(del.Method.Module.Assembly); - - engine.Dispose(); - return weak; - } -} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagSourceTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagSourceTests.cs deleted file mode 100644 index c5bbd6aa..00000000 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/VirtualTagSourceTests.cs +++ /dev/null @@ -1,138 +0,0 @@ -using Serilog; -using Shouldly; -using Xunit; -using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Core.Scripting; -using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; - -namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests; - -/// -/// Verifies the IReadable + ISubscribable adapter that DriverNodeManager dispatches -/// to for NodeSource.Virtual per ADR-002. Key contract: OPC UA clients see virtual -/// tags via the same capability interfaces as driver tags, so dispatch stays -/// source-agnostic. -/// -[Trait("Category", "Unit")] -public sealed class VirtualTagSourceTests -{ - private static (VirtualTagEngine engine, VirtualTagSource source, FakeUpstream up) Build() - { - var up = new FakeUpstream(); - up.Set("In", 10); - var logger = new LoggerConfiguration().CreateLogger(); - var engine = new VirtualTagEngine(up, new ScriptLoggerFactory(logger), logger); - engine.Load([new VirtualTagDefinition( - "Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value * 2;""")]); - return (engine, new VirtualTagSource(engine), up); - } - - /// Verifies that ReadAsync returns cached engine values. - [Fact] - public async Task ReadAsync_returns_engine_cached_values() - { - var (engine, source, _) = Build(); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var results = await source.ReadAsync(["Out"], TestContext.Current.CancellationToken); - results.Count.ShouldBe(1); - results[0].Value.ShouldBe(20); - results[0].StatusCode.ShouldBe(0u); - engine.Dispose(); - } - - /// Verifies that unknown paths return Bad status quality. - [Fact] - public async Task ReadAsync_unknown_path_returns_Bad_quality() - { - var (engine, source, _) = Build(); - var results = await source.ReadAsync(["NoSuchTag"], TestContext.Current.CancellationToken); - results[0].StatusCode.ShouldBe(0x80340000u); - engine.Dispose(); - } - - /// Verifies that subscribe fires the initial data callback immediately. - [Fact] - public async Task SubscribeAsync_fires_initial_data_callback() - { - var (engine, source, _) = Build(); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var events = new List(); - source.OnDataChange += (_, e) => events.Add(e); - - var handle = await source.SubscribeAsync(["Out"], TimeSpan.FromMilliseconds(100), - TestContext.Current.CancellationToken); - handle.ShouldNotBeNull(); - - // Per OPC UA convention, initial-data callback fires on subscribe. - events.Count.ShouldBeGreaterThanOrEqualTo(1); - events[0].FullReference.ShouldBe("Out"); - events[0].Snapshot.Value.ShouldBe(20); - - await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); - engine.Dispose(); - } - - /// Verifies that subscription fires on upstream changes via engine cascade. - [Fact] - public async Task SubscribeAsync_fires_on_upstream_change_via_engine_cascade() - { - var (engine, source, up) = Build(); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var events = new List(); - source.OnDataChange += (_, e) => events.Add(e); - var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero, - TestContext.Current.CancellationToken); - - var initialCount = events.Count; - up.Push("In", 50); - - // Wait for the cascade. - var deadline = DateTime.UtcNow.AddSeconds(2); - while (DateTime.UtcNow < deadline && events.Count <= initialCount) await Task.Delay(25); - - events.Count.ShouldBeGreaterThan(initialCount); - events[^1].Snapshot.Value.ShouldBe(100); - - await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); - engine.Dispose(); - } - - /// Verifies that unsubscribe stops further event emissions. - [Fact] - public async Task UnsubscribeAsync_stops_further_events() - { - var (engine, source, up) = Build(); - await engine.EvaluateAllAsync(TestContext.Current.CancellationToken); - - var events = new List(); - source.OnDataChange += (_, e) => events.Add(e); - var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero, - TestContext.Current.CancellationToken); - - await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); - var countAfterUnsub = events.Count; - - up.Push("In", 99); - await Task.Delay(200); - - events.Count.ShouldBe(countAfterUnsub, "Unsubscribe must stop OnDataChange emissions"); - engine.Dispose(); - } - - /// Verifies that null arguments are rejected. - [Fact] - public async Task Null_arguments_rejected() - { - var (engine, source, _) = Build(); - await Should.ThrowAsync(async () => - await source.ReadAsync(null!, TestContext.Current.CancellationToken)); - await Should.ThrowAsync(async () => - await source.SubscribeAsync(null!, TimeSpan.Zero, TestContext.Current.CancellationToken)); - await Should.ThrowAsync(async () => - await source.UnsubscribeAsync(null!, TestContext.Current.CancellationToken)); - engine.Dispose(); - } -} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests.csproj b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests.csproj index 4be9782d..d692711b 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests.csproj +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests.csproj @@ -7,11 +7,6 @@ false true ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests - - $(NoWarn);OTOPCUA0001 diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs index b656fcd5..87e2bf0e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; public sealed class LastValueCacheTests { // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value - // observed yet" (mirrored by CalculationDriver, VirtualTagEngine, FOCAS, and + // observed yet" (mirrored by CalculationDriver, FOCAS, and // OtOpcUaNodeManager for freshly-materialised / not-yet-evaluated nodes). MQTT is // subscribe-first, so an unseen RawPath is in exactly that state until its first publish. private const uint BadWaitingForInitialData = 0x80320000u; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierProvisioningTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierProvisioningTests.cs index 9e6141a3..44aea3af 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierProvisioningTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierProvisioningTests.cs @@ -146,6 +146,46 @@ public sealed class AddressSpaceApplierProvisioningTests prov.Seen[0].TagName.ShouldBe("Pump1.Good"); } + /// #600.1: a CHANGED raw tag that flips to historized, or whose historian name / data type + /// changes, is provisioned; an unchanged-historization edit and an alarm-bearing tag are not. The + /// added "Control" tag is the positive control so the absences are not a vacuous no-dispatch. + [Fact] + public async Task Apply_provisions_changed_tags_whose_historization_flips_or_name_changes() + { + var prov = new CapturingProvisioner(); + var applier = new AddressSpaceApplier(NullOpcUaAddressSpaceSink.Instance, NullLogger.Instance, prov); + + var flipOff = NonHistorizedTag("Flip", "Float32") with { TagId = "t-flip", NodeId = "Plant/Flip" }; + var flipOn = flipOff with { IsHistorized = true }; + var renOld = HistorizedTag("Ren", "Old.Name", "Float32") with { TagId = "t-ren" }; + var renNew = renOld with { HistorianTagname = "New.Name" }; + var sameOld = HistorizedTag("Same", "Same.Name", "Float32") with { TagId = "t-same", Writable = false }; + var sameNew = sameOld with { Writable = true }; // non-historian edit + var alarmOld = NonHistorizedTag("Alm", "Float32") with { TagId = "t-alm" }; + var alarmNew = alarmOld with + { + IsHistorized = true, + Alarm = new EquipmentTagAlarmInfo("AlarmCondition", 500), + }; + + var plan = PlanWithAddedTags(HistorizedTag("Control", "Control.Name", "Float32")) with + { + ChangedRawTags = new[] + { + new AddressSpacePlan.RawTagDelta(flipOff, flipOn), + new AddressSpacePlan.RawTagDelta(renOld, renNew), + new AddressSpacePlan.RawTagDelta(sameOld, sameNew), + new AddressSpacePlan.RawTagDelta(alarmOld, alarmNew), + }, + }; + + applier.Apply(plan); + + await prov.Called.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + prov.Seen.Select(r => r.TagName).ShouldBe( + new[] { "Control.Name", "Plant/Flip", "New.Name" }, ignoreOrder: true); + } + /// Capturing that records every log line and signals a /// the first time a "historian provisioning completed" tally is /// emitted, so the fire-and-forget continuation's log can be awaited deterministically. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs index 32531012..91f655da 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs @@ -122,14 +122,18 @@ public sealed class AlarmHistorianRegistrationTests /// [Theory] // Neither key: an unpaired node. Its rows exist nowhere else. - [InlineData(null, null, true)] + [InlineData(null, null, null, true)] // The dialing half of a pair — gated, because its partner holds the same rows. - [InlineData("http://peer:9001", null, false)] + [InlineData("http://peer:9001", null, null, false)] // The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would - // wrongly class it as unpaired and leave it permanently ungated. - [InlineData(null, "9001", false)] + // wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared + // ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port. + [InlineData(null, "9001", "k", false)] + // A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the + // queue is shared with nobody and the node must keep draining (#603: this used to be gated). + [InlineData(null, "9001", null, true)] public async Task Only_an_unreplicated_node_ignores_the_role_view( - string? peerAddress, string? syncListenPort, bool expectDrain) + string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain) { var tempDir = Path.Combine(Path.GetTempPath(), "otopcua-alarmhist-test-" + Guid.NewGuid().ToString("N")); Directory.CreateDirectory(tempDir); @@ -143,6 +147,7 @@ public sealed class AlarmHistorianRegistrationTests ["LocalDb:Path"] = dbPath, ["LocalDb:Replication:PeerAddress"] = peerAddress, ["LocalDb:SyncListenPort"] = syncListenPort, + ["LocalDb:Replication:ApiKey"] = apiKey, }); var services = BaseServices(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs new file mode 100644 index 00000000..0ff37296 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs @@ -0,0 +1,32 @@ +using Microsoft.Extensions.Configuration; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; + +/// #603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate. +public sealed class LocalDbReplicationPolicyTests +{ + private static IConfiguration Config(params (string Key, string? Value)[] pairs) => + new ConfigurationBuilder() + .AddInMemoryCollection(pairs.Select(p => new KeyValuePair(p.Key, p.Value))) + .Build(); + + /// PeerAddress OR ApiKey counts; the listen port alone and blank values do not. + [Theory] + [InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)] + [InlineData("LocalDb:Replication:ApiKey", "k", true)] + [InlineData("LocalDb:SyncListenPort", "9001", false)] + [InlineData("LocalDb:Replication:ApiKey", " ", false)] + public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected) + { + LocalDbReplicationPolicy.IsConfigured(Config((key, value))).ShouldBe(expected); + } + + /// No replication keys at all is an unpaired node. + [Fact] + public void IsConfigured_is_false_for_an_empty_configuration() + { + LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse(); + } +}