diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index ef029efb..fdfc8cf6 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -512,16 +512,16 @@ same pass (a UNS reference variable, say) are visible through the index with no ⚠️ **The value fan is now a single message.** It sends one `DriverInstanceActor.AttributeValueBatchPublished` carrying one item per RawPath, where it previously sent -one `AttributeValuePublished` per ref — 5 000 messages onto the mux that also carries the node's steady +one per-value message per ref — 5 000 messages onto the mux that also carries the node's steady value ingress, every time one driver flapped. Delivery is unchanged: the mux filters the batch per item against the same registered-interest map the per-value receive consulted, so only the framing moved. The **alarm** fan is indexed but deliberately not batched — one driver ref can back several condition NodeIds, so it still emits one `AlarmQualityUpdate` per condition. -**Consequence worth knowing: `AttributeValuePublished` now has no production sender at all.** This fan was -its last one — everything on the subscription path was already wrapped in `AttributeValueBatchPublished`. -`DependencyMuxActor` still registers the per-value receive, but in production it is dead code; it stays -for tests and direct senders. +**Consequence: the per-value `AttributeValuePublished` record is deleted.** This fan was its last +production sender (everything on the subscription path was already wrapped in +`AttributeValueBatchPublished`), so the record and `DependencyMuxActor`'s per-value receive were removed +(2026-10-22). A single value is a one-item batch; tests build one with `ValuePublishes.Single`. #### The third fan: the value nodes (archreview 03/S15) diff --git a/docs/ReadWriteOperations.md b/docs/ReadWriteOperations.md index c1f8b03a..e519f5e4 100644 --- a/docs/ReadWriteOperations.md +++ b/docs/ReadWriteOperations.md @@ -39,8 +39,8 @@ This was not always true. Until #519 the publish path projected the driver's `ui `Good`/`Uncertain`/`Bad` enum in `DriverInstanceActor`, and `OtOpcUaNodeManager` re-expanded `Bad` to a bare `StatusCodes.Bad` (`0x80000000`). Every driver's diagnostic intent survived in logs only — including the 16 status constants #497 had just corrected across six drivers. The value path now -carries the raw `uint` end to end (`AttributeValuePublished` → `AttributeValueUpdate` → -`IOpcUaAddressSpaceSink.WriteValue`), and severity classification is a helper +carries the raw `uint` end to end (`AttributeValueBatchPublished` → `AttributeValueBatchUpdate` → +`IOpcUaAddressSpaceSink.WriteValues`), and severity classification is a helper (`OpcUaStatus.IsGood`/`IsUncertain`/`IsBad`) rather than a lossy projection. **Writing a status code in engine code:** use a named constant on diff --git a/docs/ScriptEditor.md b/docs/ScriptEditor.md index eb6985e2..e039653c 100644 --- a/docs/ScriptEditor.md +++ b/docs/ScriptEditor.md @@ -198,10 +198,32 @@ scripted-alarm predicates) runs scripts through the shared `ScriptRunGuard`: 2 s default) quarantines immediately, through the same path (Error log, metric `reason=stuck`, Bad). A short overrun (a young orphan) stays a plain refusal and the script resumes once the orphan completes. +- **Cooperative cancellation (`ctx.CancellationToken`).** The guard cancels the run + token when a run times out, and scripts see it as `ctx.CancellationToken`. A + long-running loop that polls `ctx.CancellationToken.IsCancellationRequested` (or calls + `ctx.CancellationToken.ThrowIfCancellationRequested()`) ends its own abandoned run + and frees the slot, instead of holding a pool thread until it returns. A script that + ignores the token is still an orphan. The editor completes the member, and + `CancellationToken` is on the sandbox allow-list. - **Allocation budget.** A single run allocating more than `Scripting:MaxAllocBytesPerRun` (64 MB) is quarantined. This is sampled after the run returns, so a run that OOMs the process first is not stopped — that needs an out-of-process runner. +- **Configuration keys and their ranges.** All three knobs are node-wide (the + `Scripting` appsettings section, shared by virtual tags, Calculation tags and + scripted alarms) and are clamped when bound, with one startup Warning per + clamped value: + + | Key | Default | Range after the clamp | + |---|---|---| + | `Scripting:MaxAllocBytesPerRun` | `67108864` (64 MB) | `0` (or negative) disables the budget; a positive value below `1048576` (1 MB) becomes 1 MB | + | `Scripting:QuarantineAfterConsecutiveTimeouts` | `3` | `1`–`1000` | + | `Scripting:StuckOrphanQuarantineFactor` | `10` | `0` (or negative) disables the rule; above `1000` becomes `1000`; `NaN` becomes the default | + + A Calculation driver may override `maxAllocBytesPerRun` and + `quarantineAfterConsecutiveTimeouts` in its own driver config. Those keys are + layered on the node values and clamped to the same ranges, and an out-of-range + value logs a Warning naming the driver. The stuck factor has no per-driver key. - **Lifting.** A quarantine lifts when the script source changes or on redeploy, never by itself. - **Script-log rate limit.** The cluster-wide script-log fan-out admits 20 entries @@ -243,7 +265,7 @@ prefix. The concrete implementation `ScriptTagCatalog` reads `Tag` + | Virtual tag | Leaf `Name` (best-effort) | **Why only resolvable keys?** The live runtime resolves a `ctx.GetTag("X")` -literal against `DriverInstanceActor.AttributeValuePublished.FullReference`, +literal against `DriverInstanceActor.AttributeValueItem.FullReference`, which is the `FullName` field from `Tag.TagConfig` (see `ScriptTagCatalog.ExtractFullNameFromTagConfig` + `EquipmentNodeWalker.ExtractFullName`). The former UNS-path cascade engine (`Core.VirtualTags.VirtualTagEngine`, keyed by the diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index 183fcef0..2b7dc28a 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -93,7 +93,7 @@ Virtual-tag nodes are materialised by `AddressSpaceApplier` and fed by `VirtualT `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. +- **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes the items of each `DriverInstanceActor.AttributeValueBatchPublished` to the `VirtualTagActor` instances that declared interest in those tag refs. Each `VirtualTagActor` holds the in-memory per-tag dependency map; the `IVirtualTagEvaluator` (`RoslynVirtualTagEvaluator`) receives the dependency snapshot 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. 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 @@ -104,7 +104,7 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro - `AddressSpaceComposer.Compose` emits `DriverInstancePlan` / `ScriptedAlarmPlan` records; the driver-role `DriverHostActor` spawns one `VirtualTagActor` per virtual-tag expression and one `ScriptedAlarmActor` per scripted alarm. - `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. +- `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) receives every `DriverInstanceActor.AttributeValueBatchPublished` and routes each item to the `VirtualTagActor` instances that registered interest in that tag ref. `ITagUpstreamSource` and `IHistoryWriter` remain available as `Core.VirtualTags` ports. @@ -125,7 +125,7 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro - `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/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.Runtime/VirtualTags/DependencyMuxActor.cs` — routes `DriverInstanceActor.AttributeValueBatchPublished` items to interested `VirtualTagActor` subscribers - `src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs` — production `IVirtualTagEvaluator` binding; holds a per-source `CompiledScriptCache` - `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs` — pure data composer: config-DB entities → `AddressSpaceComposition` (UNS topology + raw subtree + driver/alarm plans) - `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs` — applies the composed plan into the SDK node manager diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDependencyConsumer.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDependencyConsumer.cs index e3261ceb..bbe5adba 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDependencyConsumer.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDependencyConsumer.cs @@ -11,7 +11,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; /// /// The mechanism mirrors how a VirtualTagActor registers interest on the /// DependencyMuxActor: the mux already receives every driver's -/// AttributeValuePublished keyed by wire-ref (under v3, RawPath), so a driver that +/// published values (batched, one item per value) keyed by wire-ref (under v3, RawPath), so a driver that /// implements this interface simply piggy-backs on that fan-out. A calc tag's computed output /// re-enters the mux via the ordinary driver publish path, so calc-of-calc chains work with no /// extra machinery — which is exactly why the deploy-time cycle gate is mandatory. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ScriptContext.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ScriptContext.cs index 6a73c622..b4fdf3cd 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ScriptContext.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ScriptContext.cs @@ -80,6 +80,50 @@ public abstract class ScriptContext /// public abstract ILogger Logger { get; } + /// + /// The current run's cancellation token. The run guard cancels it when the run exceeds its wall-clock + /// budget, and it is linked to the caller's token, so it is also cancelled when the caller cancels (host + /// shutdown, an actor stopping mid-run). A long-running script can poll ctx.CancellationToken.IsCancellationRequested (or call + /// ThrowIfCancellationRequested()) and end its own abandoned run instead of holding a pool thread + /// until it returns. Cooperative only: a script that never looks at it still runs to completion as an + /// orphan. outside a run. + /// + /// + /// Scoped per run through an , not stored on the context, because a context + /// object can be reused across runs (the scripted-alarm engine keeps one per alarm). An abandoned orphan + /// therefore keeps seeing its own cancelled token even after a newer run starts on the same context. + /// + public CancellationToken CancellationToken => s_runToken.Value; + + private static readonly AsyncLocal s_runToken = new(); + + /// + /// Makes the script-visible on the calling + /// execution context until the returned scope is disposed. Called by ScriptEvaluator.RunAsync + /// around the compiled delegate. + /// + /// The run token. + /// A scope that restores the previous token. + internal static RunTokenScope EnterRun(CancellationToken token) + { + var previous = s_runToken.Value; + s_runToken.Value = token; + return new RunTokenScope(previous); + } + + /// Restores the token that was current before . + internal readonly struct RunTokenScope : IDisposable + { + private readonly CancellationToken _previous; + + /// Captures the token to restore. + /// The token current before the run. + public RunTokenScope(CancellationToken previous) => _previous = previous; + + /// + public void Dispose() => s_runToken.Value = _previous; + } + /// /// Deadband helper — returns true when differs /// from by more than . diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions.csproj b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions.csproj index f9305356..0f07e95c 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions.csproj +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions/ZB.MOM.WW.OtOpcUa.Core.Scripting.Abstractions.csproj @@ -16,4 +16,10 @@ + + + + + diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs index c7ac8280..a33fba6f 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs @@ -185,7 +185,13 @@ public sealed class ScriptEvaluator : IDisposable // TResult directly). We surface a Task only to keep the existing // RunAsync contract consumers depend on. TimedScriptEvaluator wraps this in // Task.Run so a long-running script still honours its wall-clock budget. - var result = _func(globals); + // The run token is the script-visible ctx.CancellationToken for exactly this call. ScriptRunGuard passes + // the token it cancels on timeout, so a polling script can end its own abandoned run. + TResult result; + using (ScriptContext.EnterRun(ct)) + { + result = _func(globals); + } return Task.FromResult(result); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs index de1afee2..7efa4026 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs @@ -96,9 +96,11 @@ public sealed class ScriptRunGuard /// Where the quarantine Error line goes; null ⇒ Serilog's global logger. public ScriptRunGuard(ScriptingOptions? options = null, ILogger? logger = null) { - options ??= new ScriptingOptions(); + // Clamped here as well as at the bind site, because options also arrive from a Calculation driver's own + // config and from tests. An unclamped factor made the refusal path's timeout * factor overflow. + options = (options ?? new ScriptingOptions()).Sanitized(out _); _maxAllocBytesPerRun = options.MaxAllocBytesPerRun; - _quarantineThreshold = Math.Max(1, options.QuarantineAfterConsecutiveTimeouts); + _quarantineThreshold = options.QuarantineAfterConsecutiveTimeouts; _stuckFactor = options.StuckOrphanQuarantineFactor; _logger = logger ?? Log.Logger; } @@ -197,7 +199,11 @@ public sealed class ScriptRunGuard if (_stuckFactor > 0 && Volatile.Read(ref state.RunStartedTimestamp) is var startedAt and not 0) { var running = Stopwatch.GetElapsedTime(startedAt); - var threshold = timeout * _stuckFactor; + // Saturating: even the clamped maximum factor times a huge timeout must not overflow TimeSpan. + var thresholdTicks = timeout.Ticks * _stuckFactor; + var threshold = thresholdTicks >= TimeSpan.MaxValue.Ticks + ? TimeSpan.MaxValue + : TimeSpan.FromTicks((long)thresholdTicks); if (running > threshold) { Quarantine(key, state, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs index 296d862c..167e124f 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs @@ -53,4 +53,85 @@ public sealed class ScriptingOptions /// refusal. Zero or negative disables the rule. Default 10. /// public double StuckOrphanQuarantineFactor { get; set; } = DefaultStuckOrphanQuarantineFactor; + + /// Upper bound applies to . + public const double MaxStuckOrphanQuarantineFactor = 1000; + + /// Upper bound applies to . + public const int MaxQuarantineAfterConsecutiveTimeouts = 1000; + + /// + /// Lower bound applies to a positive : 1 MB. A + /// smaller budget would quarantine practically every script on its first run. + /// + public const long MinMaxAllocBytesPerRun = 1024L * 1024; + + /// + /// Returns a copy with every numeric knob pulled into a sane range. The bind site calls it and logs + /// . calls it again, so options that reach the + /// guard by another route (a Calculation driver's own config, a test) are clamped too. + /// + /// : NaN becomes the default, a negative value becomes 0 + /// (disabled), and a value above becomes that maximum. + /// Unclamped, an absurd factor made timeout * factor throw on + /// the first refused run. + /// : into [1, ]. + /// : a negative value becomes 0 (disabled), and a positive value + /// below becomes that minimum. + /// + /// + /// One line per value that was out of range, naming its key. A negative value that + /// already meant "disabled" is normalised without a line. + /// A new, clamped instance. This instance is not modified. + public ScriptingOptions Sanitized(out IReadOnlyList adjustments) + { + var notes = new List(); + + var factor = StuckOrphanQuarantineFactor; + if (double.IsNaN(factor)) + { + notes.Add($"{SectionName}:{nameof(StuckOrphanQuarantineFactor)} is NaN; using the default {DefaultStuckOrphanQuarantineFactor}."); + factor = DefaultStuckOrphanQuarantineFactor; + } + else if (factor < 0) + { + factor = 0; + } + else if (factor > MaxStuckOrphanQuarantineFactor) + { + notes.Add($"{SectionName}:{nameof(StuckOrphanQuarantineFactor)} {factor} is above the maximum; clamped to {MaxStuckOrphanQuarantineFactor}."); + factor = MaxStuckOrphanQuarantineFactor; + } + + var threshold = QuarantineAfterConsecutiveTimeouts; + if (threshold < 1) + { + notes.Add($"{SectionName}:{nameof(QuarantineAfterConsecutiveTimeouts)} {threshold} is below 1; clamped to 1."); + threshold = 1; + } + else if (threshold > MaxQuarantineAfterConsecutiveTimeouts) + { + notes.Add($"{SectionName}:{nameof(QuarantineAfterConsecutiveTimeouts)} {threshold} is above the maximum; clamped to {MaxQuarantineAfterConsecutiveTimeouts}."); + threshold = MaxQuarantineAfterConsecutiveTimeouts; + } + + var alloc = MaxAllocBytesPerRun; + if (alloc < 0) + { + alloc = 0; + } + else if (alloc > 0 && alloc < MinMaxAllocBytesPerRun) + { + notes.Add($"{SectionName}:{nameof(MaxAllocBytesPerRun)} {alloc} is below the minimum; clamped to {MinMaxAllocBytesPerRun}."); + alloc = MinMaxAllocBytesPerRun; + } + + adjustments = notes; + return new ScriptingOptions + { + MaxAllocBytesPerRun = alloc, + QuarantineAfterConsecutiveTimeouts = threshold, + StuckOrphanQuarantineFactor = factor, + }; + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs index 1a589d65..2c417de1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs @@ -38,13 +38,16 @@ public static class CalculationDriverFactoryExtensions /// The driver factory registry to register with. /// Optional logger factory for per-instance diagnostics loggers. /// Optional root script logger (fans failure entries onto the script-logs topic). + /// The node-wide Scripting options every Calculation driver inherits; a key + /// authored in a driver's own config overrides that one knob. Null ⇒ the compiled defaults. public static void Register( DriverFactoryRegistry registry, ILoggerFactory? loggerFactory = null, - ScriptRootLogger? scriptRoot = null) + ScriptRootLogger? scriptRoot = null, + ScriptingOptions? scripting = null) { ArgumentNullException.ThrowIfNull(registry); - registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory, scriptRoot)); + registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory, scriptRoot, scripting)); } /// Public overload for the Server-side bootstrapper + test consumers. @@ -63,6 +66,18 @@ public static class CalculationDriverFactoryExtensions public static CalculationDriver CreateInstance( string driverInstanceId, string driverConfigJson, ILoggerFactory? loggerFactory, ScriptRootLogger? scriptRoot) + => CreateInstance(driverInstanceId, driverConfigJson, loggerFactory, scriptRoot, nodeScripting: null); + + /// The full overload: also takes the node-wide Scripting options the driver inherits. + /// The unique identifier for the driver instance. + /// The merged driver config JSON (carries the RawTags array). + /// Optional logger factory for per-instance diagnostics loggers. + /// Optional root script logger for failure fan-out. + /// The node-wide run-guard options; null ⇒ the compiled defaults. + /// The constructed . + public static CalculationDriver CreateInstance( + string driverInstanceId, string driverConfigJson, + ILoggerFactory? loggerFactory, ScriptRootLogger? scriptRoot, ScriptingOptions? nodeScripting) { ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId); ArgumentException.ThrowIfNullOrWhiteSpace(driverConfigJson); @@ -71,15 +86,14 @@ public static class CalculationDriverFactoryExtensions ?? throw new InvalidOperationException( $"Calculation driver config for '{driverInstanceId}' deserialised to null"); - // Run-guard knobs are optional and default to the node-wide ScriptingOptions defaults; only an - // explicitly authored key overrides one (so a config carrying neither binds Scripting = null). - ScriptingOptions? scripting = null; - if (dto.MaxAllocBytesPerRun is not null || dto.QuarantineAfterConsecutiveTimeouts is not null) - { - scripting = new ScriptingOptions(); - if (dto.MaxAllocBytesPerRun is { } alloc) scripting.MaxAllocBytesPerRun = alloc; - if (dto.QuarantineAfterConsecutiveTimeouts is { } n) scripting.QuarantineAfterConsecutiveTimeouts = n; - } + var scripting = ResolveScripting( + dto.MaxAllocBytesPerRun, dto.QuarantineAfterConsecutiveTimeouts, nodeScripting, out var adjustments); + var logger = loggerFactory?.CreateLogger(); + // An authored value is never changed silently: the run guard would clamp it anyway, without a word. + foreach (var adjustment in adjustments) + logger?.LogWarning( + "Calculation driver {DriverInstanceId}: its run-guard config was out of range and is clamped: {Adjustment}", + driverInstanceId, adjustment); var options = new CalculationDriverOptions { @@ -89,9 +103,37 @@ public static class CalculationDriverFactoryExtensions Scripting = scripting, }; - return new CalculationDriver( - options, driverInstanceId, scriptRoot, - logger: loggerFactory?.CreateLogger()); + return new CalculationDriver(options, driverInstanceId, scriptRoot, logger: logger); + } + + /// + /// The run-guard options for one Calculation driver. The node-wide options are the base. A key authored + /// in the driver's config overrides only its own knob, on a copy, so the shared node object is never + /// written. With neither, the result is null and the guard uses its compiled defaults. + /// + /// The driver config's maxAllocBytesPerRun, or null. + /// The driver config's quarantineAfterConsecutiveTimeouts, or null. + /// The node-wide options, or null. + /// The clamps applied to the driver's OWN keys (the node options were already + /// clamped and logged at bind), for the caller to log. Empty when nothing was out of range. + /// The options to hand the driver, or null. + internal static ScriptingOptions? ResolveScripting( + long? maxAllocBytesPerRun, int? quarantineAfterConsecutiveTimeouts, ScriptingOptions? nodeScripting, + out IReadOnlyList adjustments) + { + adjustments = []; + if (maxAllocBytesPerRun is null && quarantineAfterConsecutiveTimeouts is null) return nodeScripting; + + var resolved = new ScriptingOptions(); + if (nodeScripting is not null) + { + resolved.MaxAllocBytesPerRun = nodeScripting.MaxAllocBytesPerRun; + resolved.QuarantineAfterConsecutiveTimeouts = nodeScripting.QuarantineAfterConsecutiveTimeouts; + resolved.StuckOrphanQuarantineFactor = nodeScripting.StuckOrphanQuarantineFactor; + } + if (maxAllocBytesPerRun is { } alloc) resolved.MaxAllocBytesPerRun = alloc; + if (quarantineAfterConsecutiveTimeouts is { } n) resolved.QuarantineAfterConsecutiveTimeouts = n; + return resolved.Sanitized(out adjustments); } /// The merged-config DTO the calc factory binds from. The Calculation driver has no diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/IScriptTagCatalog.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/IScriptTagCatalog.cs index 6088edb3..f4b652e2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/IScriptTagCatalog.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/IScriptTagCatalog.cs @@ -50,7 +50,7 @@ public sealed record ScriptTagInfo(string Path, string Kind, string DataType, st /// else. Verified chain: AddressSpaceComposer harvests the ctx.GetTag("…") literals /// (via EquipmentScriptPaths.ExtractDependencyRefs) into DependencyRefs; those become /// VirtualTagActor._dependencyRefs, registered with DependencyMuxActor, whose -/// _byRef map is keyed (Ordinal) by DriverInstanceActor.AttributeValuePublished.FullReference +/// _byRef map is keyed (Ordinal) by DriverInstanceActor.AttributeValueItem.FullReference /// — and in v3 a driver's wire reference is the RawPath /// (DriverHostActor._nodeIdByDriverRef is keyed (DriverInstanceId, RawPath)). /// 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 ba69b0da..23d610b7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ScriptAnalysis/ScriptAnalysisService.cs @@ -268,7 +268,16 @@ public sealed class ScriptAnalysisService var typeInfo = model.GetTypeInfo(memberAccess.Expression); var type = typeInfo.Type ?? typeInfo.ConvertedType; if (type == null) return null; - return type.GetMembers() + // The type's own members, then the INSTANCE members it inherits (stopping before System.Object, whose + // ToString/GetHashCode noise was never offered). Without the base chain, a member declared only on + // ScriptContext — ctx.CancellationToken — compiled and published but never completed. Inherited statics + // (ScriptContext.Deadband) are left out: they are not reachable through an instance receiver. This applies to + // EVERY receiver, not only ctx: a struct or enum receiver now also offers the System.ValueType / + // System.Enum instance members it inherits (CompareTo, HasFlag, ...), all of which compile. + var inherited = new List(); + for (var b = type.BaseType; b is not null && b.SpecialType != SpecialType.System_Object; b = b.BaseType) + inherited.AddRange(b.GetMembers().Where(m => !m.IsStatic)); + return type.GetMembers().Concat(inherited) .Where(m => m.CanBeReferencedByName && !m.IsImplicitlyDeclared) // NotApplicable covers members (e.g. some BCL/special members) that carry no explicit accessibility but are referenceable. .Where(m => m.DeclaredAccessibility == Accessibility.Public || m.DeclaredAccessibility == Accessibility.NotApplicable) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptingOptionsBinding.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptingOptionsBinding.cs new file mode 100644 index 00000000..1155748b --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptingOptionsBinding.cs @@ -0,0 +1,23 @@ +using Microsoft.Extensions.Configuration; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// Binds the node-wide Scripting section into and clamps it +/// (). One place, so Program.cs and its unit test bind the same +/// way and a misspelled key cannot silently fall back to the compiled defaults unnoticed. +/// +public static class ScriptingOptionsBinding +{ + /// Binds and clamps the Scripting section. + /// The host configuration. + /// One line per clamped value, for the caller to log as a Warning. + /// The clamped options; the compiled defaults when the section is absent. + public static ScriptingOptions Bind(IConfiguration configuration, out IReadOnlyList adjustments) + { + ArgumentNullException.ThrowIfNull(configuration); + var bound = configuration.GetSection(ScriptingOptions.SectionName).Get() ?? new ScriptingOptions(); + return bound.Sanitized(out adjustments); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs index 45ee0286..7cb8d869 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -68,6 +68,8 @@ public static class DriverFactoryBootstrap // The calc driver needs the root script logger so a script failure fans out onto the // script-logs topic; resolve it here (null on nodes without the script pipeline wired). var scriptRoot = sp.GetService(); + // The node-wide Scripting section (bound + clamped in Program.cs); null in test graphs ⇒ defaults. + var scripting = sp.GetService(); // The Galaxy driver resolves its Gateway.ApiKeySecretRef secret: arm through the // shared ISecretResolver (registered unconditionally by AddZbSecrets on every node). var secretResolver = sp.GetRequiredService(); @@ -84,7 +86,7 @@ public static class DriverFactoryBootstrap // failures; without this it writes to NullLogger in production. ConfigureAbTransportLogging(sp); Register(registry, loggerFactory, secretResolver, primaryRoleSource, ResolveAbCipDefaultTransport(sp), - scriptRoot); + scriptRoot, scripting); return registry; }); services.AddSingleton(sp => @@ -209,11 +211,12 @@ public static class DriverFactoryBootstrap ISecretResolver secretResolver, IDriverPrimaryRoleSource primaryRoleSource, Driver.AbCip.AbCipTransport abCipDefaultTransport, - ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger? scriptRoot = null) + ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger? scriptRoot = null, + ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptingOptions? scripting = null) { Driver.AbCip.AbCipDriverFactoryExtensions.Register(registry, loggerFactory, abCipDefaultTransport); Driver.AbLegacy.AbLegacyDriverFactoryExtensions.Register(registry, loggerFactory); - Driver.Calculation.CalculationDriverFactoryExtensions.Register(registry, loggerFactory, scriptRoot); + Driver.Calculation.CalculationDriverFactoryExtensions.Register(registry, loggerFactory, scriptRoot, scripting); Driver.FOCAS.FocasDriverFactoryExtensions.Register(registry, loggerFactory); Driver.Galaxy.GalaxyDriverFactoryExtensions.Register(registry, secretResolver, loggerFactory); Driver.Modbus.ModbusDriverFactoryExtensions.Register(registry, loggerFactory); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index b2300ab6..c3c105ab 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -365,10 +365,22 @@ if (hasDriver) // F8b — production virtual-tag evaluator (Roslyn-compiled scripts cached per expression). // Replaces the F8-default NullVirtualTagEvaluator so VirtualTagActor evaluates real user // scripts at runtime. + // Node-wide script run-guard knobs (Scripting:MaxAllocBytesPerRun / QuarantineAfterConsecutiveTimeouts / + // StuckOrphanQuarantineFactor). Bound once and clamped into sane ranges here, so an absurd value is a startup + // Warning rather than an OverflowException inside a script run. The one instance is shared by every script + // seam on the node: the virtual-tag evaluator below, the scripted-alarm engine (DriverHostActor, via + // WithOtOpcUaRuntimeActors) and every Calculation driver (DriverFactoryBootstrap). + var scriptingOptions = ZB.MOM.WW.OtOpcUa.Host.Configuration.ScriptingOptionsBinding.Bind( + builder.Configuration, out var scriptingAdjustments); + foreach (var adjustment in scriptingAdjustments) + Log.Warning("Scripting misconfiguration detected at startup: {Adjustment}", adjustment); + builder.Services.AddSingleton(scriptingOptions); + builder.Services.AddSingleton(sp => new RoslynVirtualTagEvaluator( sp.GetRequiredService().CreateLogger(), - sp.GetRequiredService())); + sp.GetRequiredService(), + scripting: sp.GetRequiredService())); builder.Services.AddSingleton(sp => sp.GetRequiredService()); // Script-log fan-out (Layer 0). The DPS publisher resolves the ActorSystem lazily so it never diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 76c82ebf..fd29c0b8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; /// and the ONLY value ingress from driver children ( wraps even a /// single data change in an AttributeValueBatchPublished). /// fabricates a batch of the same shape straight to the dependency mux, so the host neither receives nor -/// sends the per-value AttributeValuePublished any more. +/// sends a per-value message any more (the per-value record is deleted). /// public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedStash { @@ -202,6 +202,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private readonly IHistoryWriter _historyWriter; private readonly IActorRef? _virtualTagHostOverride; private readonly ScriptRootLogger? _scriptRootLogger; + + /// Node-wide script run-guard options for the scripted-alarm engine; null ⇒ compiled defaults. + private readonly ScriptingOptions? _scripting; + + /// + /// Driver children this host has stopped (, an operator Restart) and is watching + /// until they terminate. A value burst from one of them is DROPPED in : + /// Context.Stop is asynchronous, so the old child keeps publishing for the rest of the apply + /// handler and its own stop, and forwarding those bursts after would + /// re-retain a removed driver's last Good in the dependency mux indefinitely. Akka delivers + /// after the last message the watched child sent here, so removing the entry + /// there bounds the set without ever dropping a live incarnation's values. + /// + private readonly HashSet _retiredChildren = new(); private readonly IActorRef? _scriptedAlarmHostOverride; private readonly ILoggingAdapter _log = Context.GetLogger(); @@ -264,6 +278,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// from . private IActorRef? _scriptedAlarmHost; + /// The scripted-alarm plans last sent to , re-sent when it reports a + /// supervision restart (). Null until the + /// first apply. + private IReadOnlyList? _lastScriptedAlarmPlans; + private RevisionHash? _currentRevision; /// @@ -641,8 +660,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// VirtualTag host instead of spawning a real child, so tests /// can intercept the message. Null in /// production (the real host is spawned). - /// Retained for constructor-signature stability; unused since the - /// EfAlarmConditionStateStore fallback that consumed it was retired. Defaults to null. /// Optional root script logger required to spawn the ScriptedAlarm /// host (the engine + its script logging hang off it). When null the ScriptedAlarm host is left /// unspawned — the graceful dev/None-deployment path. @@ -694,6 +711,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// capture to a mesh-partner Primary the cluster failure detector reports unreachable. /// Where the OPC UA HistoryUpdate verdict is published for the node manager's /// HistoryUpdate Primary gate. Null on nodes that wire no redundancy manager. + /// The node-wide Scripting run-guard options handed to the scripted-alarm + /// engine. Null ⇒ the compiled defaults. /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -707,7 +726,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS IVirtualTagEvaluator? virtualTagEvaluator = null, IHistoryWriter? historyWriter = null, IActorRef? virtualTagHostOverride = null, - ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, IDriverCapabilityInvokerFactory? invokerFactory = null, @@ -728,7 +746,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Func? sharesAlarmQueueWithLocal = null, TimeSpan? applyOutcomeTimeout = null, bool captureWhilePrimaryUnreachable = true, - IHistoryUpdateRoleView? historyUpdateRoleView = null) => + IHistoryUpdateRoleView? historyUpdateRoleView = null, + ScriptingOptions? scripting = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -737,11 +756,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Akka.Actor.Props.Create(() => new DriverHostActor( dbFactory, localNode, ackRouter, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, - loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, + scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay, - sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView)); + sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView, + scripting)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -760,8 +780,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// for historized VirtualTag results; defaults to . /// Test seam: when supplied, used as the VirtualTag host /// instead of spawning a real child. - /// Retained for constructor-signature stability; unused since the - /// EfAlarmConditionStateStore fallback that consumed it was retired. Defaults to null. /// Optional root script logger required to spawn the ScriptedAlarm /// host; when null the host is left unspawned. /// Test seam: when supplied, used as the ScriptedAlarm host @@ -830,6 +848,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// ( — the data-plane verdict, but closed until the first /// redundancy snapshot) is published for the node manager's HistoryUpdate Primary gate. Null on nodes that wire no /// redundancy manager. + /// The node-wide Scripting run-guard options handed to the scripted-alarm + /// engine. Null ⇒ the compiled defaults. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -842,7 +862,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS IVirtualTagEvaluator? virtualTagEvaluator = null, IHistoryWriter? historyWriter = null, IActorRef? virtualTagHostOverride = null, - ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, IDriverCapabilityInvokerFactory? invokerFactory = null, @@ -863,8 +882,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Func? sharesAlarmQueueWithLocal = null, TimeSpan? applyOutcomeTimeout = null, bool captureWhilePrimaryUnreachable = true, - IHistoryUpdateRoleView? historyUpdateRoleView = null) + IHistoryUpdateRoleView? historyUpdateRoleView = null, + ScriptingOptions? scripting = null) { + _scripting = scripting; _captureWhilePrimaryUnreachable = captureWhilePrimaryUnreachable; _historyUpdateRoleView = historyUpdateRoleView; _applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout; @@ -1146,8 +1167,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// null and ApplyScriptedAlarms becomes a no-op. The engine is built around a fresh /// + the wired (the /// replicated LocalDbAlarmConditionStateStore on a driver-role node); when no store was - /// wired the host is skipped. The host (spawned as a child) owns + disposes the engine in its - /// PostStop, so it stops with the driver host. + /// wired the host is skipped. The host is spawned through the factory Props, so EACH + /// incarnation (a supervision restart included) builds, owns and disposes its own engine; it stops + /// with the driver host. /// private void SpawnScriptedAlarmHost() { @@ -1178,13 +1200,23 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS return; } + // Factory Props: every incarnation of the host (a supervision restart included) builds and owns a fresh + // (upstream, engine) pair, so a restart never reuses an engine the previous incarnation disposed or left + // half-loaded. The closure captures locals only; it runs in the child's constructor. var store = _alarmStateStore; - var upstream = new DependencyMuxTagUpstreamSource(); - var engine = new ScriptedAlarmEngine( - upstream, store, new ScriptLoggerFactory(_scriptRootLogger.Logger), _scriptRootLogger.Logger); + var scriptLogger = _scriptRootLogger.Logger; + var scripting = _scripting; _scriptedAlarmHost = Context.ActorOf( ScriptedAlarmHostActor.Props( - _opcUaPublishActor, _dependencyMux, upstream, engine, _localNode, telemetryHub: _telemetryHub), + _opcUaPublishActor, _dependencyMux, + () => + { + var upstream = new DependencyMuxTagUpstreamSource(); + var engine = new ScriptedAlarmEngine( + upstream, store, new ScriptLoggerFactory(scriptLogger), scriptLogger, scripting: scripting); + return (upstream, engine); + }, + _localNode, telemetryHub: _telemetryHub), "scripted-alarm-host"); } @@ -1500,6 +1532,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); + Receive(OnRetiredChildTerminated); + Receive(_ => HandleScriptedAlarmHostRestarted()); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleCommsRecoveryAudit); @@ -1562,6 +1596,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); + Receive(OnRetiredChildTerminated); + Receive(_ => HandleScriptedAlarmHostRestarted()); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleCommsRecoveryAudit); @@ -1606,6 +1642,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg) { + // A burst from an incarnation this host already stopped: forwarding it would undo the SourceRemoved + // eviction (see _retiredChildren). Its ConnectivityChanged is NOT filtered — the respawn path relies on + // that Bad fan to cover the gap. + if (_retiredChildren.Contains(Sender)) + { + _log.Debug("DriverHost {Node}: dropped a {Count}-item value burst from a stopped incarnation of {Id}", + _localNode, msg.Items.Count, msg.DriverInstanceId); + return; + } + // One envelope per burst; the mux filters per item (it registered a batch receive). _dependencyMux?.Tell(msg); @@ -2501,6 +2547,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(m => _log.Debug("DriverHost {Node}: dropping a {Count}-item value burst from {Id} while Stale (no address space)", _localNode, m.Items.Count, m.DriverInstanceId)); + Receive(OnRetiredChildTerminated); + Receive(_ => HandleScriptedAlarmHostRestarted()); Receive(m => _log.Debug("DriverHost {Node}: dropping a native alarm transition ({Cond}) from {Id} while Stale", _localNode, m.Args.ConditionId, m.DriverInstanceId)); @@ -3652,6 +3700,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // (re)loads its engine + re-registers mux interest for the union of dependency refs. Covers // both the fresh-apply and bootstrap-restore paths (both call this method); the Stale-recovery // path deliberately does not, matching driver + VirtualTag recovery. + _lastScriptedAlarmPlans = composition.EquipmentScriptedAlarms; _scriptedAlarmHost?.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(composition.EquipmentScriptedAlarms)); if (composition.EquipmentScriptedAlarms.Count > 0) { @@ -3816,8 +3865,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS { if (!_children.TryGetValue(driverInstanceId, out var entry)) return; StopDependencyAdapter(driverInstanceId); + RetireChild(entry.Actor); Context.Stop(entry.Actor); _children.Remove(driverInstanceId); + EvictRetainedValues(driverInstanceId); // The instance left the plan (StopChild is only reached via plan.ToStop), so its cached // Health/Resilience snapshots must not replay to new telemetry subscribers. Release the shared // resilience state FIRST: the resilience publisher re-emits the tracker's snapshots every tick, so @@ -3836,6 +3887,56 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); } + /// + /// Tells the dependency mux to drop every retained value this driver instance produced, so a stopped + /// driver's last Good is never replayed to a later replay-opting subscriber as if it were current. Sent + /// wherever a driver OBJECT ends: a deploy removal, the stop half of a config-change respawn + /// () and an operator Restart (a new driver object). Deliberately NOT sent on a + /// supervision restart (): that re-uses the same driver, whose last value is + /// still its current one. Sent before any respawn, on the same host→mux pair the new incarnation's + /// values travel, so the eviction is ordered ahead of them. + /// + /// + /// The stopped child keeps publishing until it actually terminates, and for a REMOVED driver nothing + /// would ever overwrite a value re-retained after this eviction (it never publishes again, and its + /// PostStop Bad fan resolves no refs once the apply has rebuilt the index without it). So the caller + /// also s it, and drops every burst from a + /// retired child. + /// + /// The driver instance that ended. + private void EvictRetainedValues(string driverInstanceId) => + _dependencyMux?.Tell(new DependencyMuxActor.SourceRemoved(driverInstanceId)); + + /// Marks a child this host is about to stop as retired and watches it until it terminates. + /// The child being stopped. + private void RetireChild(IActorRef child) + { + _retiredChildren.Add(child); + Context.Watch(child); + } + + /// + /// The scripted-alarm host was restarted by supervision and has an empty engine: re-send the last plans so + /// its inputs and operator commands work again without waiting for the next deploy. Only the CURRENT host's + /// signal is honoured. + /// + private void HandleScriptedAlarmHostRestarted() + { + if (_scriptedAlarmHost is null || !Sender.Equals(_scriptedAlarmHost)) return; + if (_lastScriptedAlarmPlans is null) + { + _log.Debug("DriverHost {Node}: ScriptedAlarm host restarted before any apply; nothing to re-send", _localNode); + return; + } + _log.Warning("DriverHost {Node}: ScriptedAlarm host was restarted by supervision; re-applying {Count} scripted alarm(s)", + _localNode, _lastScriptedAlarmPlans.Count); + _scriptedAlarmHost.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(_lastScriptedAlarmPlans)); + } + + /// A retired child has terminated; nothing more can arrive from it. + /// The termination notice. + private void OnRetiredChildTerminated(Terminated msg) => _retiredChildren.Remove(msg.ActorRef); + private static string ActorNameFor(string driverInstanceId, long generation) { // Akka actor names cannot contain '/', ':', or whitespace. Mangle defensively. The monotonic @@ -3881,8 +3982,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Stop the existing child actor — DriverInstanceActor.PostStop calls ShutdownAsync. StopDependencyAdapter(msg.DriverInstanceId); + RetireChild(entry.Actor); Context.Stop(entry.Actor); _children.Remove(msg.DriverInstanceId); + // An operator Restart builds a NEW driver object, so the old one's retained values are evicted like a + // removal's; the respawned driver republishes its own. + EvictRetainedValues(msg.DriverInstanceId); // The old incarnation's health must not read as the new one's until it publishes its own. _healthPublisher.Forget(msg.DriverInstanceId); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index c1ad189b..287e0763 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -108,21 +108,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// state — a no-op when already Reconnecting or Connecting. /// public sealed record ForceReconnect; - /// One driver-published value. No production sender remains — the subscription path - /// batches every data change into an , and the comms-loss fan - /// (DriverHostActor.FanCommsLossToDependencyMux), which was the last sender to fabricate this - /// message, now fabricates a batch too. The dependency mux still accepts it for tests and direct senders. + /// One value within an . /// is the driver's full OPC UA status code including its sub-code /// — not a severity bucket. Classify it with and friends rather - /// than comparing it to . - public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); - /// One value within an . + /// than comparing it to . /// A struct deliberately: an AttributeValueItem[] then holds its items inline, so a burst of /// N values costs one array allocation instead of one array plus N boxes. Indexed access through /// does not box. public readonly record struct AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); - /// Batched form of — every changed attribute of ONE - /// driver production burst in one message. The parent fans the whole batch to the mux and the OPC UA + /// The ONLY value message a driver child sends: every changed attribute of ONE driver production + /// burst in one message (a single data change is a one-item batch). The parent fans the whole batch to the mux and the OPC UA /// publish path in one hop each. Status codes ride verbatim. public sealed record AttributeValueBatchPublished(string DriverInstanceId, IReadOnlyList Items); private sealed record DataChangeForward(string FullReference, DataValueSnapshot Snapshot); @@ -132,7 +127,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private sealed record DataChangeBatchForward(IReadOnlyList Items); /// Published to the parent whenever the subscribed driver (an ) fires /// . The parent () projects + routes it - /// to the materialised Part 9 condition. Parallels . + /// to the materialised Part 9 condition. Parallels . public sealed record AttributeAlarmPublished(string DriverInstanceId, AlarmEventArgs Args); private sealed record NativeAlarmRaised(AlarmEventArgs Args); /// Self-sent on Connected entry / when alarm refs are (re)pushed, to establish the native-alarm diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index 210a84ca..60f05729 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -86,6 +86,21 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// The desired Equipment-namespace scripted-alarm plans. public sealed record ApplyScriptedAlarms(IReadOnlyList Plans); + /// + /// Told to the parent from . A supervision restart is otherwise invisible to the + /// parent, and the new incarnation has an EMPTY engine (factory Props) and no mux interest, so until the + /// next its inputs stop and every operator command is dropped as not + /// owned. The parent (DriverHostActor) answers by re-sending the last plans it applied, mirroring + /// how a driver child's ReprovisionRequested is answered. + /// + public sealed class ScriptedAlarmHostRestarted + { + /// The single instance. + public static readonly ScriptedAlarmHostRestarted Instance = new(); + + private ScriptedAlarmHostRestarted() { } + } + /// Marshals an engine emission off the engine's worker thread onto the actor thread. /// Carries the the engine raised on OnEvent. private sealed record EngineEmission(ScriptedAlarmEvent Event); @@ -384,9 +399,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor // subscriptions exist, so any DependencyValueChanged the mux delivers will be observed by the // engine. RegisterInterest replaces a subscriber's prior interest set (see DependencyMuxActor), // so a re-Apply with a fresh union simply supersedes the old one — no explicit unregister needed. - // ReplayRetained: the mux replays the last value it saw for each ref, so a fresh engine (a - // supervision restart builds one around an EMPTY upstream) converges immediately instead of - // waiting for every input to change again. + // ReplayRetained: the mux replays the last value it saw for each ref, so a freshly loaded engine + // converges immediately instead of waiting for every input to change again. That includes the engine + // a supervision restart builds around an EMPTY upstream, but only once the parent has re-sent + // ApplyScriptedAlarms in answer to ScriptedAlarmHostRestarted (PostRestart); nothing reloads it before. _mux?.Tell(new DependencyMuxActor.RegisterInterest(msg.DepRefs, Self, ReplayRetained: true)); // Prune the upstream's retained snapshots to the loaded set (R02:L3): a path no alarm references // any more will never be pushed again, so its last value is dead weight for the process lifetime. @@ -791,6 +807,15 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor base.PreRestart(reason, message); } + /// + /// Tells the parent so it re-sends the last + /// : the new incarnation has nothing loaded and no mux interest. + protected override void PostRestart(Exception reason) + { + base.PostRestart(reason); + Context.Parent.Tell(ScriptedAlarmHostRestarted.Instance); + } + /// protected override void PostStop() { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index e6e58419..ea327b42 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -609,7 +609,6 @@ public static class ServiceCollectionExtensions healthPublisher: healthPublisher, virtualTagEvaluator: virtualTagEvaluator, historyWriter: historyWriter, - loggerFactory: loggerFactory, scriptRootLogger: scriptRootLogger, invokerFactory: invokerFactory, deploymentArtifactCache: deploymentArtifactCache, @@ -638,7 +637,10 @@ public static class ServiceCollectionExtensions sharesAlarmQueueWithLocal: BuildAlarmQueuePeerPredicate( resolver.GetService(), roleInfo, replicationPeerHost), // The HistoryUpdate verdict (closed until the first redundancy snapshot). - historyUpdateRoleView: resolver.GetService()), + historyUpdateRoleView: resolver.GetService(), + // The node-wide Scripting section (bound + clamped in Program.cs) for the scripted-alarm + // engine. Unbound in test graphs ⇒ null ⇒ the compiled defaults. + scripting: resolver.GetService()), DriverHostActorName); registry.Register(driverHost); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs index 40e9bf19..51174b9e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs @@ -15,11 +15,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// DriverHostActor forwards driver-child value publishes to this mux (one mux per driver-role /// node). The ONLY driver ingress is the batched /// — one envelope per driver -/// production burst, which the mux filters per item. The per-value -/// receive has no production sender left -/// at all: the comms-loss fan (DriverHostActor.FanCommsLossToDependencyMux) was its last -/// one and now fabricates an AttributeValueBatchPublished instead. The receive is kept for -/// tests and direct senders — treat it as dead in production. +/// production burst, which the mux filters per item. The comms-loss fan +/// (DriverHostActor.FanCommsLossToDependencyMux) fabricates one too. There is no per-value +/// ingress: a single value is a one-item batch (tests build one with ValuePublishes.Single). /// /// The mux emits ONLY batches, so a subscriber's /// batch receive is its complete hot path. Every subscriber keeps a per-value @@ -52,11 +50,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// Source removal. The mux records which driver instance last published each retained ref; /// evicts every value that driver produced, so a removed driver's last /// Good is never replayed as current. A comms-loss fan (a Bad batch) already overwrites the retained -/// value with Bad. ⚠️ As of fix round 1 no production actor sends yet — -/// DriverHostActor (outside the scripting lane) must Tell it when it stops / removes a driver -/// child. Until then a removed driver's last value can still be replayed to a replay-opting -/// subscriber that re-registers for that ref while another replay-opting subscriber holds it, or -/// within the grace window. +/// value with Bad. DriverHostActor sends wherever a driver object +/// ends (a deploy removal, the stop half of a config-change respawn, an operator Restart), and never +/// on a supervision restart, which re-uses the same driver. /// /// public sealed class DependencyMuxActor : ReceiveActor, IWithTimers @@ -154,7 +150,6 @@ public sealed class DependencyMuxActor : ReceiveActor, IWithTimers Receive(msg => RemoveSubscriber(msg.Subscriber)); Receive(OnSourceRemoved); Receive(_ => OnSweep()); - Receive(OnAttributeValuePublished); Receive(OnAttributeValueBatchPublished); Receive(msg => RemoveSubscriber(msg.ActorRef)); } @@ -265,30 +260,6 @@ public sealed class DependencyMuxActor : ReceiveActor, IWithTimers _log.Debug("DependencyMux: source {Driver} removed — evicted {Count} retained value(s)", msg.DriverInstanceId, evicted); } - private void OnAttributeValuePublished(DriverInstanceActor.AttributeValuePublished msg) - { - if (!_byRef.TryGetValue(msg.FullReference, out var entry)) - { - // Nothing is registered for (or retaining) this ref — drop. Common in normal operation; the address - // space carries thousands of tags and only a fraction feed virtual-tag expressions. - return; - } - var dep = new VirtualTagActor.DependencyValueChanged(msg.FullReference, msg.Value, msg.TimestampUtc, msg.StatusCode); - if (entry.Retains) - { - entry.Last = dep; - entry.LastSource = msg.DriverInstanceId; - } - foreach (var sub in entry.Subscribers) - { - // A single-item batch, so the mux emits exactly ONE message shape and every subscriber needs - // only the batch receive on its hot path. One list per subscriber: batch ownership transfers - // to the receiver, so a shared list must never cross mailboxes. Nothing in production reaches - // this path (see the class doc) — the extra allocation is noise on a test-only seam. - sub.Tell(new VirtualTagActor.DependencyValueBatch(new[] { dep })); - } - } - private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg) { // Nothing registered and nothing retained — skip the whole burst rather than probing an empty map per item. diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptCancellationTokenTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptCancellationTokenTests.cs new file mode 100644 index 00000000..a465d2d4 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptCancellationTokenTests.cs @@ -0,0 +1,67 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; + +/// +/// ctx.CancellationToken — the run token, visible to scripts. already +/// cancels the token it hands when a run times out; +/// a script that polls it can now end its own abandoned run instead of burning a pool thread until it +/// returns. Cooperative only: a script that ignores it is still an orphan. +/// +[Trait("Category", "Unit")] +public sealed class ScriptCancellationTokenTests +{ + /// Spins until cancelled, but ends by itself after 10 s so a broken implementation cannot pin a + /// pool thread past the test. + private const string CooperativeSpin = + """ + var end = System.DateTime.UtcNow.AddSeconds(10); + while (!ctx.CancellationToken.IsCancellationRequested && System.DateTime.UtcNow < end) { } + return ctx.CancellationToken.IsCancellationRequested ? -1 : 1; + """; + + [Fact] + public async Task A_script_sees_an_uncancelled_token_on_a_normal_run() + { + using var eval = ScriptEvaluator.Compile( + "return ctx.CancellationToken.CanBeCanceled && !ctx.CancellationToken.IsCancellationRequested;"); + using var cts = new CancellationTokenSource(); + + (await eval.RunAsync(new FakeScriptContext(), cts.Token)).ShouldBeTrue(); + } + + [Fact] + public async Task Outside_a_run_the_token_is_None() + { + using var eval = ScriptEvaluator.Compile("return 1;"); + var ctx = new FakeScriptContext(); + (await eval.RunAsync(ctx)).ShouldBe(1); + + ctx.CancellationToken.ShouldBe(CancellationToken.None, "the token is scoped to the run, not left on the context"); + } + + [Fact] + public async Task A_timed_out_run_whose_script_polls_the_token_ends_its_own_orphan() + { + var guard = new ScriptRunGuard(); + using var eval = ScriptEvaluator.Compile(CooperativeSpin); + var ct = TestContext.Current.CancellationToken; + + await Should.ThrowAsync( + () => guard.RunAsync("k", CooperativeSpin, eval, new FakeScriptContext(), TimeSpan.FromMilliseconds(100), ct)); + + // The guard cancelled the run token on timeout; the script saw it and returned, which frees the slot well + // inside its own 10 s fallback. Without ctx.CancellationToken the orphan would hold it for the full 10 s. + var deadline = DateTime.UtcNow.AddSeconds(3); + while (guard.InFlightCount != 0 && DateTime.UtcNow < deadline) + await Task.Delay(20, ct); + guard.InFlightCount.ShouldBe(0, "the cooperative script ended its abandoned run"); + + // Positive control: the slot is free, so the next run is dispatched rather than refused. Same source + // identity on purpose: a different one would reset the key and dispatch even behind a live orphan. + using var fast = ScriptEvaluator.Compile("return 5;"); + (await guard.RunAsync("k", CooperativeSpin, fast, new FakeScriptContext(), TimeSpan.FromSeconds(5), ct)) + .ShouldBe(5); + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptingOptionsTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptingOptionsTests.cs new file mode 100644 index 00000000..c702854a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptingOptionsTests.cs @@ -0,0 +1,109 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; + +/// +/// — the bind-time clamp of the Scripting section's numeric +/// knobs. An absurd value must be pulled into a sane range at bind time rather than surfacing later as an +/// inside a script run (timeout * StuckOrphanQuarantineFactor). +/// +[Trait("Category", "Unit")] +public sealed class ScriptingOptionsTests +{ + [Fact] + public void Defaults_pass_through_unchanged_with_no_adjustments() + { + var sanitized = new ScriptingOptions().Sanitized(out var adjustments); + + adjustments.ShouldBeEmpty(); + sanitized.MaxAllocBytesPerRun.ShouldBe(ScriptingOptions.DefaultMaxAllocBytesPerRun); + sanitized.QuarantineAfterConsecutiveTimeouts.ShouldBe(ScriptingOptions.DefaultQuarantineAfterConsecutiveTimeouts); + sanitized.StuckOrphanQuarantineFactor.ShouldBe(ScriptingOptions.DefaultStuckOrphanQuarantineFactor); + } + + [Theory] + [InlineData(double.MaxValue, ScriptingOptions.MaxStuckOrphanQuarantineFactor)] + [InlineData(double.PositiveInfinity, ScriptingOptions.MaxStuckOrphanQuarantineFactor)] + [InlineData(1e300, ScriptingOptions.MaxStuckOrphanQuarantineFactor)] + [InlineData(double.NaN, ScriptingOptions.DefaultStuckOrphanQuarantineFactor)] + [InlineData(-5, 0)] + [InlineData(double.NegativeInfinity, 0)] + [InlineData(0, 0)] + [InlineData(2.5, 2.5)] + public void StuckOrphanQuarantineFactor_is_clamped(double configured, double expected) + { + var sanitized = new ScriptingOptions { StuckOrphanQuarantineFactor = configured }.Sanitized(out var adjustments); + + sanitized.StuckOrphanQuarantineFactor.ShouldBe(expected); + // Negative already means "disabled" and is documented as such; normalising it to 0 is not worth a warning. + if (!double.IsNaN(configured) && configured >= 0 && configured <= ScriptingOptions.MaxStuckOrphanQuarantineFactor) + adjustments.ShouldBeEmpty(); + } + + [Theory] + [InlineData(0, 1)] + [InlineData(-3, 1)] + [InlineData(int.MaxValue, ScriptingOptions.MaxQuarantineAfterConsecutiveTimeouts)] + [InlineData(5, 5)] + public void QuarantineAfterConsecutiveTimeouts_is_clamped(int configured, int expected) => + new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = configured }.Sanitized(out _) + .QuarantineAfterConsecutiveTimeouts.ShouldBe(expected); + + [Theory] + [InlineData(-1L, 0L)] + [InlineData(0L, 0L)] + [InlineData(1024L, ScriptingOptions.MinMaxAllocBytesPerRun)] + [InlineData(long.MaxValue, long.MaxValue)] + public void MaxAllocBytesPerRun_is_clamped(long configured, long expected) => + new ScriptingOptions { MaxAllocBytesPerRun = configured }.Sanitized(out _) + .MaxAllocBytesPerRun.ShouldBe(expected); + + [Fact] + public void An_out_of_range_value_is_reported_as_an_adjustment_naming_the_key() + { + _ = new ScriptingOptions { StuckOrphanQuarantineFactor = 1e300 }.Sanitized(out var adjustments); + + adjustments.ShouldHaveSingleItem().ShouldContain("Scripting:StuckOrphanQuarantineFactor"); + } + + [Fact] + public void Sanitized_returns_a_copy_and_leaves_the_source_untouched() + { + var source = new ScriptingOptions { StuckOrphanQuarantineFactor = 1e300 }; + + var sanitized = source.Sanitized(out _); + + sanitized.ShouldNotBeSameAs(source); + source.StuckOrphanQuarantineFactor.ShouldBe(1e300); + } + + /// + /// The guard takes options from more places than the bind (the Calculation driver's per-driver DTO, tests), + /// so it must not overflow either: an absurd factor reaching the guard directly is clamped, and a refusal + /// behind a young orphan is still a plain refusal rather than an . + /// + [Fact] + public async Task An_absurd_factor_reaching_the_guard_directly_does_not_overflow_on_a_refusal() + { + var guard = new ScriptRunGuard(new ScriptingOptions { StuckOrphanQuarantineFactor = double.MaxValue }); + const string spin = + """ + var end = System.DateTime.UtcNow.AddSeconds(3); + while (System.DateTime.UtcNow < end) { } + return 1; + """; + using var eval = ScriptEvaluator.Compile(spin); + var ct = TestContext.Current.CancellationToken; + var timeout = TimeSpan.FromMilliseconds(60); + + await Should.ThrowAsync( + () => guard.RunAsync("k", spin, eval, new FakeScriptContext(), timeout, ct)); + // The refusal path evaluates timeout * factor. Unclamped, TimeSpan * double.MaxValue throws OverflowException. + await Should.ThrowAsync( + () => guard.RunAsync("k", spin, eval, new FakeScriptContext(), timeout, ct)); + // A huge timeout times the clamped maximum factor must not overflow either. + await Should.ThrowAsync( + () => guard.RunAsync("k", spin, eval, new FakeScriptContext(), TimeSpan.FromDays(365 * 1000), ct)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationScriptingOptionsResolutionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationScriptingOptionsResolutionTests.cs new file mode 100644 index 00000000..5d2a86b7 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationScriptingOptionsResolutionTests.cs @@ -0,0 +1,69 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests; + +/// +/// The Calculation driver's run-guard options: the node-wide Scripting section is the base, and a key +/// authored in the driver's own config overrides only that knob. Before the section was bound, a Calculation +/// driver always ran on the compiled defaults whatever the node configured. +/// +[Trait("Category", "Unit")] +public sealed class CalculationScriptingOptionsResolutionTests +{ + private static readonly ScriptingOptions Node = new() + { + MaxAllocBytesPerRun = 8L * 1024 * 1024, + QuarantineAfterConsecutiveTimeouts = 7, + StuckOrphanQuarantineFactor = 4, + }; + + [Fact] + public void No_driver_keys_and_no_node_options_resolves_null_so_the_guard_uses_its_defaults() => + CalculationDriverFactoryExtensions.ResolveScripting(null, null, null, out _).ShouldBeNull(); + + [Fact] + public void No_driver_keys_inherits_every_node_knob() + { + var resolved = CalculationDriverFactoryExtensions.ResolveScripting(null, null, Node, out _).ShouldNotBeNull(); + + resolved.MaxAllocBytesPerRun.ShouldBe(Node.MaxAllocBytesPerRun); + resolved.QuarantineAfterConsecutiveTimeouts.ShouldBe(7); + resolved.StuckOrphanQuarantineFactor.ShouldBe(4); + } + + [Fact] + public void A_driver_key_overrides_only_its_own_knob_and_the_node_object_is_not_mutated() + { + var resolved = CalculationDriverFactoryExtensions.ResolveScripting(null, 2, Node, out var adjustments).ShouldNotBeNull(); + + resolved.QuarantineAfterConsecutiveTimeouts.ShouldBe(2); + resolved.MaxAllocBytesPerRun.ShouldBe(Node.MaxAllocBytesPerRun, "the unauthored key keeps the node value"); + resolved.StuckOrphanQuarantineFactor.ShouldBe(4, "the stuck factor has no driver key; it is node-wide"); + resolved.ShouldNotBeSameAs(Node); + Node.QuarantineAfterConsecutiveTimeouts.ShouldBe(7); + adjustments.ShouldBeEmpty(); + } + + /// Review m-2: an out-of-range value authored in the driver's own config is clamped AND reported, so + /// the factory can log it rather than the guard changing it silently. + [Fact] + public void An_out_of_range_driver_key_is_clamped_and_reported() + { + var resolved = CalculationDriverFactoryExtensions.ResolveScripting(1024L, 0, Node, out var adjustments).ShouldNotBeNull(); + + resolved.MaxAllocBytesPerRun.ShouldBe(ScriptingOptions.MinMaxAllocBytesPerRun); + resolved.QuarantineAfterConsecutiveTimeouts.ShouldBe(1); + adjustments.Count.ShouldBe(2); + } + + [Fact] + public void A_driver_key_with_no_node_options_layers_on_the_compiled_defaults() + { + var resolved = CalculationDriverFactoryExtensions.ResolveScripting(0L, null, null, out _).ShouldNotBeNull(); + + resolved.MaxAllocBytesPerRun.ShouldBe(0); + resolved.QuarantineAfterConsecutiveTimeouts.ShouldBe(ScriptingOptions.DefaultQuarantineAfterConsecutiveTimeouts); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Authorization/PageAuthorizationGuardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Authorization/PageAuthorizationGuardTests.cs index b58c9500..231c61cc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Authorization/PageAuthorizationGuardTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Authorization/PageAuthorizationGuardTests.cs @@ -57,10 +57,11 @@ public class PageAuthorizationGuardTests // AddSecretsAuthorization(). ── [typeof(SecretsAdmin)] = global::ZB.MOM.WW.Secrets.Ui.SecretsAuthorization.ManagePolicy, - // ── AuthenticatedRead (16): dashboards, lists, live tails (read-only) + the dev ContextMenu demo ── + // ── AuthenticatedRead (17): dashboards, lists, live tails (read-only) + the dev ContextMenu demo ── [typeof(Home)] = AdminUiPolicies.AuthenticatedRead, [typeof(global::ZB.MOM.WW.OtOpcUa.AdminUI.Components.Pages.Dev.ContextMenuDemo)] = AdminUiPolicies.AuthenticatedRead, [typeof(Fleet)] = AdminUiPolicies.AuthenticatedRead, + [typeof(global::ZB.MOM.WW.OtOpcUa.AdminUI.Components.Pages.FleetDiagnostics)] = AdminUiPolicies.AuthenticatedRead, [typeof(global::ZB.MOM.WW.OtOpcUa.AdminUI.Components.Pages.Hosts)] = AdminUiPolicies.AuthenticatedRead, [typeof(Alerts)] = AdminUiPolicies.AuthenticatedRead, [typeof(ScriptLog)] = AdminUiPolicies.AuthenticatedRead, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/CompletionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/CompletionTests.cs index 63b319da..be424a2a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/CompletionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/CompletionTests.cs @@ -18,6 +18,7 @@ public sealed class CompletionTests labels.ShouldContain("SetVirtualTag"); labels.ShouldContain("Now"); labels.ShouldContain("Logger"); + labels.ShouldContain("CancellationToken"); } [Fact] public async Task Scope_completion_includes_the_ctx_local() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/DiagnoseTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/DiagnoseTests.cs index 94784a45..27c8679e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/DiagnoseTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ScriptAnalysis/DiagnoseTests.cs @@ -48,6 +48,7 @@ public sealed class DiagnoseTests [InlineData("""var x = 41; System.Threading.Interlocked.Increment(ref x); return x;""")] [InlineData("""var x = 42; return System.Threading.Volatile.Read(ref x);""")] [InlineData("""object o = 42; return o.GetType().IsValueType ? 1 : 0;""")] + [InlineData("""ctx.CancellationToken.ThrowIfCancellationRequested(); return ctx.CancellationToken.IsCancellationRequested ? 1 : 0;""")] public void Allow_listed_forms_are_not_flagged_in_the_editor(string code) => Diag(code).ShouldBeEmpty(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptingOptionsBindingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptingOptionsBindingTests.cs new file mode 100644 index 00000000..f717d6c5 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptingOptionsBindingTests.cs @@ -0,0 +1,64 @@ +using Microsoft.Extensions.Configuration; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration; + +/// +/// Review m-1: the Scripting section binds by the documented key names and is clamped at bind. Each +/// configured value differs from the compiled default, so a misspelled key (which the binder would silently +/// ignore, falling back to the default) goes red. +/// +public sealed class ScriptingOptionsBindingTests +{ + private static IConfiguration Config(Dictionary values) => + new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + + [Fact] + public void Every_documented_key_binds_to_its_property() + { + var bound = ScriptingOptionsBinding.Bind(Config(new() + { + ["Scripting:MaxAllocBytesPerRun"] = "8388608", + ["Scripting:QuarantineAfterConsecutiveTimeouts"] = "7", + ["Scripting:StuckOrphanQuarantineFactor"] = "4.5", + // Sibling keys owned by the script-log pipeline must not disturb the bind. + ["Scripting:LogFilePath"] = "logs/x-.log", + ["Scripting:LogTopicMinLevel"] = "Warning", + }), out var adjustments); + + adjustments.ShouldBeEmpty(); + bound.MaxAllocBytesPerRun.ShouldBe(8L * 1024 * 1024); + bound.QuarantineAfterConsecutiveTimeouts.ShouldBe(7); + bound.StuckOrphanQuarantineFactor.ShouldBe(4.5); + } + + [Fact] + public void Out_of_range_values_are_clamped_at_bind_and_reported() + { + var bound = ScriptingOptionsBinding.Bind(Config(new() + { + ["Scripting:StuckOrphanQuarantineFactor"] = "1e300", + ["Scripting:MaxAllocBytesPerRun"] = "1024", + ["Scripting:QuarantineAfterConsecutiveTimeouts"] = "0", + }), out var adjustments); + + bound.StuckOrphanQuarantineFactor.ShouldBe(ScriptingOptions.MaxStuckOrphanQuarantineFactor); + bound.MaxAllocBytesPerRun.ShouldBe(ScriptingOptions.MinMaxAllocBytesPerRun); + bound.QuarantineAfterConsecutiveTimeouts.ShouldBe(1); + adjustments.Count.ShouldBe(3); + } + + [Fact] + public void An_absent_section_yields_the_compiled_defaults() + { + var bound = ScriptingOptionsBinding.Bind(Config(new()), out var adjustments); + + adjustments.ShouldBeEmpty(); + bound.MaxAllocBytesPerRun.ShouldBe(ScriptingOptions.DefaultMaxAllocBytesPerRun); + bound.QuarantineAfterConsecutiveTimeouts.ShouldBe(ScriptingOptions.DefaultQuarantineAfterConsecutiveTimeouts); + bound.StuckOrphanQuarantineFactor.ShouldBe(ScriptingOptions.DefaultStuckOrphanQuarantineFactor); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorScriptedAlarmRestartTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorScriptedAlarmRestartTests.cs new file mode 100644 index 00000000..2ff682f2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorScriptedAlarmRestartTests.cs @@ -0,0 +1,103 @@ +using System.Text.Json; +using Akka.Actor; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review I-2: under the factory Props a supervision restart of the scripted-alarm host builds a fresh, EMPTY +/// engine. Nothing re-sent ApplyScriptedAlarms, so until the next deploy the host had no mux interest +/// (inputs stopped) and its ownership filter dropped every operator ack/shelve as "not owned". The host now +/// tells its parent from PostRestart, +/// and re-sends the last plans it applied. This test never re-applies by hand. +/// +public sealed class DriverHostActorScriptedAlarmRestartTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-alarmhost-restart-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('d', 64)); + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + + private static DeploymentId SeedAlarmDeployment(Microsoft.EntityFrameworkCore.IDbContextFactory db) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + ScriptedAlarms = new[] + { + new + { + ScriptedAlarmId = "alm-1", EquipmentId = "eq-1", Name = "HighTemp", AlarmType = "AlarmCondition", + Severity = 800, MessageTemplate = "hot", PredicateScriptId = "s-1", + HistorizeToAveva = false, Retain = true, Enabled = true, + }, + }, + Scripts = new[] { new { ScriptId = "s-1", SourceCode = "return (int)ctx.GetTag(\"M.T\").Value > 90;" } }, + }); + var id = DeploymentId.NewId(); + using var ctx = db.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = id.Value, RevisionHash = RevA.Value, Status = DeploymentStatus.Sealed, + CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + return id; + } + + [Fact] + public void After_a_supervision_restart_the_alarm_host_is_re_applied_and_inputs_and_acks_work_again() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedAlarmDeployment(db); + var publish = this.CreateCleanOutcomePublishProbe(); + var mux = CreateTestProbe(); + var coordinator = CreateTestProbe(); + var host = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + dependencyMux: mux.Ref, + opcUaPublishActor: publish.Ref, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: CreateTestProbe().Ref, + scriptRootLogger: new ScriptRootLogger(new LoggerConfiguration().CreateLogger()), + alarmStateStore: new InMemoryAlarmStateStore())); + V3ArtifactSeeder.Dispatch(host, coordinator, deploymentId, RevA); + + var first = mux.FishForMessage(r => r.TagRefs.Contains("M.T"), Timeout); + var alarmHost = first.Subscriber; + + // The crash: a batch item with a null TagId makes the upstream Push throw (the documented unguarded + // path), so supervision restarts the alarm host. NO manual re-apply follows. + alarmHost.Tell(new VirtualTagActor.DependencyValueBatch(new[] + { + new VirtualTagActor.DependencyValueChanged(null!, 1, DateTime.UtcNow), + })); + mux.FishForMessage(u => u.Subscriber.Equals(alarmHost), Timeout); + + // The parent re-applies on the restart signal: interest is registered again by the new incarnation. + mux.FishForMessage(r => r.TagRefs.Contains("M.T"), Timeout); + + // Inputs work: a dependency push activates the alarm. + alarmHost.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow)); + publish.FishForMessage( + m => m.AlarmNodeId == "alm-1" && m.State.Active && !m.State.Acknowledged, Timeout); + + // Acks work: the restarted engine owns alm-1 again, so the command is not filtered out as unowned. + alarmHost.Tell(new AlarmCommand(AlarmId: "alm-1", Operation: "Acknowledge", User: "alice", Comment: null, UnshelveAtUtc: null)); + publish.FishForMessage( + m => m.AlarmNodeId == "alm-1" && m.State.Acknowledged, Timeout) + .State.Acknowledged.ShouldBeTrue(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs new file mode 100644 index 00000000..0e7e48ac --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs @@ -0,0 +1,224 @@ +using Akka.Actor; +using Akka.TestKit; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// The dependency mux retains a ref's last value for replay-opting subscribers and evicts a driver's values on +/// . Before this, nothing in production sent that message, so a +/// driver removed by a deploy kept its last Good in the mux and replayed it to the next replay-opting +/// subscriber as if it were current. The host now sends it wherever a driver OBJECT ends (deploy removal, +/// config-change respawn, operator Restart), never on a supervision restart, and drops value bursts from the +/// child incarnations it has already stopped so a late burst cannot re-retain what was evicted. All through a +/// REAL mux and real driver children. +/// +public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-sourceremoved-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private const string Path1 = "Plant/Modbus/dev1/speed"; + private const string Path2 = "Plant/Modbus2/dev2/flow"; + private static readonly (string, string, string, string, string?)[] NoRefs = []; + + private static readonly (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] BothDrivers = + [ + ("DI-1", "Modbus", "dev1", "speed", "Double", false), + ("DI-2", "Modbus2", "dev2", "flow", "Double", false), + ]; + + private IActorRef _host = ActorRefs.Nobody; + private IActorRef _mux = ActorRefs.Nobody; + private TestProbe _coordinator = null!; + private TestProbe _holder = null!; + private RecordingSubscribableDriverFactory _factory = null!; + private IDbContextFactory _db = null!; + + /// Spawns a host over a real mux with two enabled drivers, waits for both children to subscribe, + /// registers a replay-opting holder (so the mux retains both refs) and publishes one value per driver. + private void ArrangeTwoDriversWithRetainedValues() + { + _db = NewInMemoryDbFactory(); + var revA = V3ArtifactSeeder.SeedV3Deployment(_db, RevA, BothDrivers, NoRefs, enabled: true); + _factory = new RecordingSubscribableDriverFactory("Modbus"); + _mux = Sys.ActorOf(DependencyMuxActor.Props()); + _coordinator = CreateTestProbe(); + _holder = CreateTestProbe(); + _host = Sys.ActorOf(DriverHostActor.Props( + _db, TestNode, _coordinator.Ref, + driverFactory: _factory, + localRoles: new HashSet { "driver" }, + dependencyMux: _mux, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: CreateTestProbe().Ref)); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revA, RevA); + AwaitAssert(() => _factory.Created.Count.ShouldBe(2), PresenceBudget); + // The spawn hands a child the PREVIOUS apply's set (empty on the first deploy) and the bulk push then + // re-subscribes with the real one; a value fired before that lands is lost, so wait for the real set. + AwaitSubscribed("DI-1", Path1); + AwaitSubscribed("DI-2", Path2); + + _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, _holder.Ref, ReplayRetained: true)); + Driver("DI-1").FireDataChange(Path1, 11.0); + Driver("DI-2").FireDataChange(Path2, 22.0); + // One fish for both: the two drivers' bursts arrive in either order, and fishing for one at a time would + // discard the other. + var pending = new HashSet(StringComparer.Ordinal) { Path1, Path2 }; + _holder.FishForMessage(b => + { + foreach (var i in b.Items) + if ((i.TagId == Path1 && Equals(i.Value, 11.0)) || (i.TagId == Path2 && Equals(i.Value, 22.0))) + pending.Remove(i.TagId); + return pending.Count == 0; + }, PresenceBudget); + } + + /// Waits until the holder has been delivered for . + private void AwaitHolder(string path, object value) => + _holder.FishForMessage( + b => b.Items.Any(i => i.TagId == path && Equals(i.Value, value)), PresenceBudget); + + private void AwaitSubscribed(string driverId, string path) => + AwaitAssert(() => (Driver(driverId).LastSubscribedRefs ?? []).ShouldContain(path), PresenceBudget); + + private RecordingSubscribableDriver Driver(string id) => _factory.Created.Last(d => d.DriverInstanceId == id); + + /// A fresh replay-opting subscriber for both refs; returns what it was replayed, keyed by ref. + private Dictionary Replayed() + { + var probe = CreateTestProbe(); + _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, probe.Ref, ReplayRetained: true)); + return probe.ExpectMsg(PresenceBudget) + .Items.ToDictionary(i => i.TagId, StringComparer.Ordinal); + } + + /// The live child actor for (named drv-{id}-g{generation}). + private IActorRef ChildOf(string driverId) + { + var probe = CreateTestProbe(); + Sys.ActorSelection(_host.Path.ToStringWithoutAddress() + "/drv-" + driverId + "-g*").Tell(new Identify(driverId), probe.Ref); + return probe.ExpectMsg(PresenceBudget).Subject.ShouldNotBeNull(); + } + + private static void ShouldNotBeTheOldGood(Dictionary replay, string path, object oldValue) + { + if (!replay.TryGetValue(path, out var item)) return; + (Equals(item.Value, oldValue) && OpcUaStatus.IsGood(item.StatusCode)) + .ShouldBeFalse($"the stopped incarnation's last Good for {path} was replayed as current"); + } + + [Fact] + public void A_driver_removed_by_a_deploy_is_no_longer_replayed_while_a_still_present_driver_is() + { + ArrangeTwoDriversWithRetainedValues(); + + // Positive control before the removal: both values replay. + Replayed().Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true); + + // Deploy B drops DI-1 (StopChild). + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + + AwaitAssert( + () => Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "DI-1 was removed: its last Good must not replay; DI-2's must"), + PresenceBudget); + } + + /// Review I-1: a burst the removed child sent after the eviction (it stops asynchronously, so it keeps + /// publishing through the rest of the apply) must be dropped by the host, not forwarded to re-retain the value. + [Fact] + public void A_late_burst_from_a_removed_child_does_not_re_retain_its_value() + { + ArrangeTwoDriversWithRetainedValues(); + var oldChild = ChildOf("DI-1"); + // Hold the removed child in its PostStop so it has NOT terminated when the stale burst lands; a terminated + // actor cannot send, and the host forgets a retired child on Terminated. + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Driver("DI-1").ShutdownGate = gate; + + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + + // The stale burst, sent AS the stopped child. + _host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild); + // Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst + // has been processed by the host by the time we look. + Driver("DI-2").FireDataChange(Path2, 23.0); + AwaitHolder(Path2, 23.0); + + Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good"); + gate.TrySetResult(); + } + + /// Review m-3: a config edit respawns the driver (stop + spawn); the OLD object's value is evicted, and + /// the NEW driver's value replays (positive control). + [Fact] + public void A_config_change_respawn_evicts_the_old_value_and_the_new_drivers_value_replays() + { + ArrangeTwoDriversWithRetainedValues(); + + // Adding a tag to DI-1 changes its merged DriverConfig, which respawns it. + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, + [.. BothDrivers, ("DI-1", "Modbus", "dev1", "temp", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget); + var respawned = Driver("DI-1"); + // The spawn hands the child the PREVIOUS apply's set, then the bulk push re-subscribes with the new one + // (which names the added tag); a value fired between the two can be lost, so wait for the final set. + AwaitAssert(() => (respawned.LastSubscribedRefs ?? []).ShouldContain("Plant/Modbus/dev1/temp"), PresenceBudget); + + ShouldNotBeTheOldGood(Replayed(), Path1, 11.0); + + respawned.FireDataChange(Path1, 33.0); + AwaitHolder(Path1, 33.0); + var replay = Replayed(); + replay[Path1].Value.ShouldBe(33.0, "the respawned driver's own value replays"); + replay[Path2].Value.ShouldBe(22.0, "the untouched driver is unaffected"); + } + + /// Review m-3: an operator Restart builds a new driver object; the old one's value is evicted and the + /// new one's replays. + [Fact] + public void An_operator_restart_evicts_the_old_value_and_the_new_drivers_value_replays() + { + ArrangeTwoDriversWithRetainedValues(); + + _host.Tell(new RestartDriver("c1", "DI-1", "tester", Guid.NewGuid())); + AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget); + var restarted = Driver("DI-1"); + AwaitSubscribed("DI-1", Path1); + + ShouldNotBeTheOldGood(Replayed(), Path1, 11.0); + + restarted.FireDataChange(Path1, 44.0); + AwaitHolder(Path1, 44.0); + Replayed()[Path1].Value.ShouldBe(44.0); + } + + [Fact] + public void A_supervision_restart_does_not_evict_the_drivers_retained_value() + { + ArrangeTwoDriversWithRetainedValues(); + var di1 = Driver("DI-1"); + + // Crash DI-1's child: Akka restarts it around the SAME driver object, and the host reprovisions it. + di1.FireMalformedDataChange(Path1); + AwaitAssert(() => di1.InitializeCount.ShouldBe(2), PresenceBudget); + _factory.Created.Count.ShouldBe(2, "a supervision restart re-uses the driver; no new one is created"); + + var replay = Replayed(); + replay.Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true, + "a restart is the same driver instance: its last value stays replayable"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs index 1c470992..1efaed5e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs @@ -110,7 +110,7 @@ public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase batch.Items.Count.ShouldBe(2); // The per-tag raises for the SAME burst produced nothing extra — no second batch, no - // AttributeValuePublished. Anything here means both handlers were attached. + // single-item batch. Anything here means both handlers were attached. parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400)); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 627a1c38..ea0cfaef 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -279,11 +279,18 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla /// public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + /// + /// When set, awaits it (the actor bounds the wait), which keeps a stopping child + /// ALIVE in its PostStop for as long as the test wants — the window in which a stopped child can still + /// have messages reaching its parent. + /// + public TaskCompletionSource? ShutdownGate { get; set; } + /// public Task ShutdownAsync(CancellationToken cancellationToken) { Interlocked.Increment(ref _shutdownCount); - return Task.CompletedTask; + return ShutdownGate?.Task ?? Task.CompletedTask; } /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs index d0582eff..4831643c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs @@ -5,8 +5,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; /// /// Builds the ONE message shape a driver child sends for its values — a /// — for tests that push a single value into the -/// dependency mux by hand. The per-value has no -/// production sender (02/U4); tests should not keep one alive either. +/// dependency mux by hand. There is no per-value message: the former per-value record had no production +/// sender and was deleted (02/U4). /// public static class ValuePublishes { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs index 10bce23b..7cccf27d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs @@ -19,7 +19,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; /// /// Adapted from the plan's Task 17: the recorder handles the REAL fan-out message the mux emits — /// (TagId/Value/TimestampUtc/StatusCode) — -/// not DriverInstanceActor.AttributeValuePublished. The typed-value / real-status behaviour is +/// not the driver-side DriverInstanceActor.AttributeValueBatchPublished. The typed-value / real-status behaviour is /// pinned in . /// public sealed class ContinuousHistorizationRecorderTests : TestKit diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs index 76ad462f..79237192 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs @@ -10,9 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags; public sealed class DependencyMuxActorTests : RuntimeActorTestBase { - /// Verifies that AttributeValuePublished messages are forwarded only to subscribers interested in the specific reference. + /// Verifies that a published value is forwarded only to subscribers interested in the specific reference. [Fact] - public void AttributeValuePublished_is_forwarded_only_to_subscribers_interested_in_that_ref() + public void A_published_value_is_forwarded_only_to_subscribers_interested_in_that_ref() { var mux = Sys.ActorOf(DependencyMuxActor.Props()); var subA = CreateTestProbe(); @@ -21,9 +21,9 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1", "tag-2" }, subA.Ref)); mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-2", "tag-3" }, subB.Ref)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-3", 30, OpcUaStatus.Good, DateTime.UtcNow)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-2", 20, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-3", 30, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-2", 20, OpcUaStatus.Good, DateTime.UtcNow)); // The mux emits ONLY batches — the per-value comms-loss ingress fans out as a single-item batch. // subA hears tag-1 + tag-2. @@ -54,7 +54,7 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase var sub = CreateTestProbe(); mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "nobody-cares", 99, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "nobody-cares", 99, OpcUaStatus.Good, DateTime.UtcNow)); sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } @@ -67,11 +67,11 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase var sub = CreateTestProbe(); mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); sub.ExpectMsg(); mux.Tell(new DependencyMuxActor.UnregisterInterest(sub.Ref)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-1", 20, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-1", 20, OpcUaStatus.Good, DateTime.UtcNow)); sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } @@ -85,10 +85,10 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref)); mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-2" }, sub.Ref)); // replaces tag-1 - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-1", 10, OpcUaStatus.Good, DateTime.UtcNow)); sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "tag-2", 20, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "tag-2", 20, OpcUaStatus.Good, DateTime.UtcNow)); sub.ExpectMsg().Items.ShouldHaveSingleItem().TagId.ShouldBe("tag-2"); } @@ -110,18 +110,18 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase // gets dropped (no subscriber for "ref-a" yet), so we re-publish each pass. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "driver-1", "ref-a", 10, OpcUaStatus.Good, DateTime.UtcNow)); parent.ExpectMsg(TimeSpan.FromMilliseconds(200)) .Value.ShouldBe(10); }, duration: PresenceBudget); // From here the actor is wired — second publish drives the sum. - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "ref-b", 32, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "ref-b", 32, OpcUaStatus.Good, DateTime.UtcNow)); parent.ExpectMsg(PresenceBudget).Value.ShouldBe(42); // Unrelated ref shouldn't fire eval. - mux.Tell(new DriverInstanceActor.AttributeValuePublished("driver-1", "ref-unrelated", 99, OpcUaStatus.Good, DateTime.UtcNow)); + mux.Tell(ValuePublishes.Single("driver-1", "ref-unrelated", 99, OpcUaStatus.Good, DateTime.UtcNow)); parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } @@ -221,7 +221,7 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase var sub = CreateTestProbe(); mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "driver-1", "tag-1", 10, OpcUaStatus.Bad, DateTime.UtcNow)); sub.ExpectMsg()