Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+23
-1
@@ -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
|
||||
|
||||
+3
-3
@@ -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
|
||||
|
||||
@@ -11,7 +11,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
/// <para>
|
||||
/// The mechanism mirrors how a <c>VirtualTagActor</c> registers interest on the
|
||||
/// <c>DependencyMuxActor</c>: the mux already receives every driver's
|
||||
/// <c>AttributeValuePublished</c> 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.
|
||||
|
||||
@@ -80,6 +80,50 @@ public abstract class ScriptContext
|
||||
/// </summary>
|
||||
public abstract ILogger Logger { get; }
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>ctx.CancellationToken.IsCancellationRequested</c> (or call
|
||||
/// <c>ThrowIfCancellationRequested()</c>) 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. <see cref="CancellationToken.None"/> outside a run.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Scoped per run through an <see cref="AsyncLocal{T}"/>, 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.
|
||||
/// </remarks>
|
||||
public CancellationToken CancellationToken => s_runToken.Value;
|
||||
|
||||
private static readonly AsyncLocal<CancellationToken> s_runToken = new();
|
||||
|
||||
/// <summary>
|
||||
/// Makes <paramref name="token"/> the script-visible <see cref="CancellationToken"/> on the calling
|
||||
/// execution context until the returned scope is disposed. Called by <c>ScriptEvaluator.RunAsync</c>
|
||||
/// around the compiled delegate.
|
||||
/// </summary>
|
||||
/// <param name="token">The run token.</param>
|
||||
/// <returns>A scope that restores the previous token.</returns>
|
||||
internal static RunTokenScope EnterRun(CancellationToken token)
|
||||
{
|
||||
var previous = s_runToken.Value;
|
||||
s_runToken.Value = token;
|
||||
return new RunTokenScope(previous);
|
||||
}
|
||||
|
||||
/// <summary>Restores the token that was current before <see cref="EnterRun"/>.</summary>
|
||||
internal readonly struct RunTokenScope : IDisposable
|
||||
{
|
||||
private readonly CancellationToken _previous;
|
||||
|
||||
/// <summary>Captures the token to restore.</summary>
|
||||
/// <param name="previous">The token current before the run.</param>
|
||||
public RunTokenScope(CancellationToken previous) => _previous = previous;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose() => s_runToken.Value = _previous;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deadband helper — returns <c>true</c> when <paramref name="current"/> differs
|
||||
/// from <paramref name="previous"/> by more than <paramref name="tolerance"/>.
|
||||
|
||||
+6
@@ -16,4 +16,10 @@
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- ScriptEvaluator scopes the script-visible run token (ScriptContext.EnterRun). Internal so a script
|
||||
cannot replace its own token. -->
|
||||
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Core.Scripting"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
@@ -185,7 +185,13 @@ public sealed class ScriptEvaluator<TContext, TResult> : IDisposable
|
||||
// TResult directly). We surface a Task<TResult> 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);
|
||||
}
|
||||
|
||||
|
||||
@@ -96,9 +96,11 @@ public sealed class ScriptRunGuard
|
||||
/// <param name="logger">Where the quarantine Error line goes; null ⇒ Serilog's global logger.</param>
|
||||
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,
|
||||
|
||||
@@ -53,4 +53,85 @@ public sealed class ScriptingOptions
|
||||
/// refusal. Zero or negative disables the rule. Default 10.
|
||||
/// </summary>
|
||||
public double StuckOrphanQuarantineFactor { get; set; } = DefaultStuckOrphanQuarantineFactor;
|
||||
|
||||
/// <summary>Upper bound <see cref="Sanitized"/> applies to <see cref="StuckOrphanQuarantineFactor"/>.</summary>
|
||||
public const double MaxStuckOrphanQuarantineFactor = 1000;
|
||||
|
||||
/// <summary>Upper bound <see cref="Sanitized"/> applies to <see cref="QuarantineAfterConsecutiveTimeouts"/>.</summary>
|
||||
public const int MaxQuarantineAfterConsecutiveTimeouts = 1000;
|
||||
|
||||
/// <summary>
|
||||
/// Lower bound <see cref="Sanitized"/> applies to a positive <see cref="MaxAllocBytesPerRun"/>: 1 MB. A
|
||||
/// smaller budget would quarantine practically every script on its first run.
|
||||
/// </summary>
|
||||
public const long MinMaxAllocBytesPerRun = 1024L * 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Returns a copy with every numeric knob pulled into a sane range. The bind site calls it and logs
|
||||
/// <paramref name="adjustments"/>. <see cref="ScriptRunGuard"/> calls it again, so options that reach the
|
||||
/// guard by another route (a Calculation driver's own config, a test) are clamped too.
|
||||
/// <list type="bullet">
|
||||
/// <item><see cref="StuckOrphanQuarantineFactor"/>: NaN becomes the default, a negative value becomes 0
|
||||
/// (disabled), and a value above <see cref="MaxStuckOrphanQuarantineFactor"/> becomes that maximum.
|
||||
/// Unclamped, an absurd factor made <c>timeout * factor</c> throw <see cref="OverflowException"/> on
|
||||
/// the first refused run.</item>
|
||||
/// <item><see cref="QuarantineAfterConsecutiveTimeouts"/>: into [1, <see cref="MaxQuarantineAfterConsecutiveTimeouts"/>].</item>
|
||||
/// <item><see cref="MaxAllocBytesPerRun"/>: a negative value becomes 0 (disabled), and a positive value
|
||||
/// below <see cref="MinMaxAllocBytesPerRun"/> becomes that minimum.</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
/// <param name="adjustments">One line per value that was out of range, naming its key. A negative value that
|
||||
/// already meant "disabled" is normalised without a line.</param>
|
||||
/// <returns>A new, clamped instance. This instance is not modified.</returns>
|
||||
public ScriptingOptions Sanitized(out IReadOnlyList<string> adjustments)
|
||||
{
|
||||
var notes = new List<string>();
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
+56
-14
@@ -38,13 +38,16 @@ public static class CalculationDriverFactoryExtensions
|
||||
/// <param name="registry">The driver factory registry to register with.</param>
|
||||
/// <param name="loggerFactory">Optional logger factory for per-instance diagnostics loggers.</param>
|
||||
/// <param name="scriptRoot">Optional root script logger (fans failure entries onto the <c>script-logs</c> topic).</param>
|
||||
/// <param name="scripting">The node-wide <c>Scripting</c> options every Calculation driver inherits; a key
|
||||
/// authored in a driver's own config overrides that one knob. Null ⇒ the compiled defaults.</param>
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>Public overload for the Server-side bootstrapper + test consumers.</summary>
|
||||
@@ -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);
|
||||
|
||||
/// <summary>The full overload: also takes the node-wide <c>Scripting</c> options the driver inherits.</summary>
|
||||
/// <param name="driverInstanceId">The unique identifier for the driver instance.</param>
|
||||
/// <param name="driverConfigJson">The merged driver config JSON (carries the <c>RawTags</c> array).</param>
|
||||
/// <param name="loggerFactory">Optional logger factory for per-instance diagnostics loggers.</param>
|
||||
/// <param name="scriptRoot">Optional root script logger for failure fan-out.</param>
|
||||
/// <param name="nodeScripting">The node-wide run-guard options; null ⇒ the compiled defaults.</param>
|
||||
/// <returns>The constructed <see cref="CalculationDriver"/>.</returns>
|
||||
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<CalculationDriver>();
|
||||
// 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<CalculationDriver>());
|
||||
return new CalculationDriver(options, driverInstanceId, scriptRoot, logger: logger);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="maxAllocBytesPerRun">The driver config's <c>maxAllocBytesPerRun</c>, or null.</param>
|
||||
/// <param name="quarantineAfterConsecutiveTimeouts">The driver config's <c>quarantineAfterConsecutiveTimeouts</c>, or null.</param>
|
||||
/// <param name="nodeScripting">The node-wide options, or null.</param>
|
||||
/// <param name="adjustments">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.</param>
|
||||
/// <returns>The options to hand the driver, or null.</returns>
|
||||
internal static ScriptingOptions? ResolveScripting(
|
||||
long? maxAllocBytesPerRun, int? quarantineAfterConsecutiveTimeouts, ScriptingOptions? nodeScripting,
|
||||
out IReadOnlyList<string> 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);
|
||||
}
|
||||
|
||||
/// <summary>The merged-config DTO the calc factory binds from. The Calculation driver has no
|
||||
|
||||
@@ -50,7 +50,7 @@ public sealed record ScriptTagInfo(string Path, string Kind, string DataType, st
|
||||
/// else. Verified chain: <c>AddressSpaceComposer</c> harvests the <c>ctx.GetTag("…")</c> literals
|
||||
/// (via <c>EquipmentScriptPaths.ExtractDependencyRefs</c>) into <c>DependencyRefs</c>; those become
|
||||
/// <c>VirtualTagActor._dependencyRefs</c>, registered with <c>DependencyMuxActor</c>, whose
|
||||
/// <c>_byRef</c> map is keyed (Ordinal) by <c>DriverInstanceActor.AttributeValuePublished.FullReference</c>
|
||||
/// <c>_byRef</c> map is keyed (Ordinal) by <c>DriverInstanceActor.AttributeValueItem.FullReference</c>
|
||||
/// — and in v3 a driver's wire reference <b>is the RawPath</b>
|
||||
/// (<c>DriverHostActor._nodeIdByDriverRef</c> is keyed <c>(DriverInstanceId, RawPath)</c>).
|
||||
/// </para>
|
||||
|
||||
@@ -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<ISymbol>();
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// Binds the node-wide <c>Scripting</c> section into <see cref="ScriptingOptions"/> and clamps it
|
||||
/// (<see cref="ScriptingOptions.Sanitized"/>). One place, so <c>Program.cs</c> and its unit test bind the same
|
||||
/// way and a misspelled key cannot silently fall back to the compiled defaults unnoticed.
|
||||
/// </summary>
|
||||
public static class ScriptingOptionsBinding
|
||||
{
|
||||
/// <summary>Binds and clamps the <c>Scripting</c> section.</summary>
|
||||
/// <param name="configuration">The host configuration.</param>
|
||||
/// <param name="adjustments">One line per clamped value, for the caller to log as a Warning.</param>
|
||||
/// <returns>The clamped options; the compiled defaults when the section is absent.</returns>
|
||||
public static ScriptingOptions Bind(IConfiguration configuration, out IReadOnlyList<string> adjustments)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(configuration);
|
||||
var bound = configuration.GetSection(ScriptingOptions.SectionName).Get<ScriptingOptions>() ?? new ScriptingOptions();
|
||||
return bound.Sanitized(out adjustments);
|
||||
}
|
||||
}
|
||||
@@ -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<ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger>();
|
||||
// The node-wide Scripting section (bound + clamped in Program.cs); null in test graphs ⇒ defaults.
|
||||
var scripting = sp.GetService<ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptingOptions>();
|
||||
// The Galaxy driver resolves its Gateway.ApiKeySecretRef secret: arm through the
|
||||
// shared ISecretResolver (registered unconditionally by AddZbSecrets on every node).
|
||||
var secretResolver = sp.GetRequiredService<ISecretResolver>();
|
||||
@@ -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<IDriverFactory>(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);
|
||||
|
||||
@@ -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<RoslynVirtualTagEvaluator>(sp =>
|
||||
new RoslynVirtualTagEvaluator(
|
||||
sp.GetRequiredService<ILoggerFactory>().CreateLogger<RoslynVirtualTagEvaluator>(),
|
||||
sp.GetRequiredService<ScriptRootLogger>()));
|
||||
sp.GetRequiredService<ScriptRootLogger>(),
|
||||
scripting: sp.GetRequiredService<ScriptingOptions>()));
|
||||
builder.Services.AddSingleton<IVirtualTagEvaluator>(sp => sp.GetRequiredService<RoslynVirtualTagEvaluator>());
|
||||
|
||||
// Script-log fan-out (Layer 0). The DPS publisher resolves the ActorSystem lazily so it never
|
||||
|
||||
@@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
/// and the ONLY value ingress from driver children (<see cref="DriverInstanceActor"/> wraps even a
|
||||
/// single data change in an <c>AttributeValueBatchPublished</c>). <see cref="FanCommsLossToDependencyMux"/>
|
||||
/// fabricates a batch of the same shape straight to the dependency mux, so the host neither receives nor
|
||||
/// sends the per-value <c>AttributeValuePublished</c> any more.
|
||||
/// sends a per-value message any more (the per-value record is deleted).
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
/// <summary>Node-wide script run-guard options for the scripted-alarm engine; null ⇒ compiled defaults.</summary>
|
||||
private readonly ScriptingOptions? _scripting;
|
||||
|
||||
/// <summary>
|
||||
/// Driver children this host has stopped (<see cref="StopChild"/>, an operator Restart) and is watching
|
||||
/// until they terminate. A value burst from one of them is DROPPED in <see cref="ForwardBatchToMux"/>:
|
||||
/// <c>Context.Stop</c> is asynchronous, so the old child keeps publishing for the rest of the apply
|
||||
/// handler and its own stop, and forwarding those bursts after <see cref="EvictRetainedValues"/> would
|
||||
/// re-retain a removed driver's last Good in the dependency mux indefinitely. Akka delivers
|
||||
/// <see cref="Terminated"/> 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.
|
||||
/// </summary>
|
||||
private readonly HashSet<IActorRef> _retiredChildren = new();
|
||||
private readonly IActorRef? _scriptedAlarmHostOverride;
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
|
||||
@@ -264,6 +278,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// from <see cref="PushDesiredSubscriptions"/>.</summary>
|
||||
private IActorRef? _scriptedAlarmHost;
|
||||
|
||||
/// <summary>The scripted-alarm plans last sent to <see cref="_scriptedAlarmHost"/>, re-sent when it reports a
|
||||
/// supervision restart (<see cref="ScriptedAlarmHostActor.ScriptedAlarmHostRestarted"/>). Null until the
|
||||
/// first apply.</summary>
|
||||
private IReadOnlyList<EquipmentScriptedAlarmPlan>? _lastScriptedAlarmPlans;
|
||||
|
||||
private RevisionHash? _currentRevision;
|
||||
|
||||
/// <summary>
|
||||
@@ -641,8 +660,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// VirtualTag host instead of spawning a real <see cref="VirtualTagHostActor"/> child, so tests
|
||||
/// can intercept the <see cref="VirtualTagHostActor.ApplyVirtualTags"/> message. Null in
|
||||
/// production (the real host is spawned).</param>
|
||||
/// <param name="loggerFactory">Retained for constructor-signature stability; unused since the
|
||||
/// <c>EfAlarmConditionStateStore</c> fallback that consumed it was retired. Defaults to null.</param>
|
||||
/// <param name="scriptRootLogger">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.</param>
|
||||
@@ -694,6 +711,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// capture to a mesh-partner Primary the cluster failure detector reports unreachable.</param>
|
||||
/// <param name="historyUpdateRoleView">Where the OPC UA HistoryUpdate verdict is published for the node manager's
|
||||
/// HistoryUpdate Primary gate. Null on nodes that wire no redundancy manager.</param>
|
||||
/// <param name="scripting">The node-wide <c>Scripting</c> run-guard options handed to the scripted-alarm
|
||||
/// engine. Null ⇒ the compiled defaults.</param>
|
||||
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
|
||||
public static Props Props(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? 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<CommonsNodeId, bool>? 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));
|
||||
|
||||
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
|
||||
/// <param name="dbFactory">Database context factory for configuration database access.</param>
|
||||
@@ -760,8 +780,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// for historized VirtualTag results; defaults to <see cref="NullHistoryWriter"/>.</param>
|
||||
/// <param name="virtualTagHostOverride">Test seam: when supplied, used as the VirtualTag host
|
||||
/// instead of spawning a real <see cref="VirtualTagHostActor"/> child.</param>
|
||||
/// <param name="loggerFactory">Retained for constructor-signature stability; unused since the
|
||||
/// <c>EfAlarmConditionStateStore</c> fallback that consumed it was retired. Defaults to null.</param>
|
||||
/// <param name="scriptRootLogger">Optional root script logger required to spawn the ScriptedAlarm
|
||||
/// host; when null the host is left unspawned.</param>
|
||||
/// <param name="scriptedAlarmHostOverride">Test seam: when supplied, used as the ScriptedAlarm host
|
||||
@@ -830,6 +848,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// (<see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/> — 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.</param>
|
||||
/// <param name="scripting">The node-wide <c>Scripting</c> run-guard options handed to the scripted-alarm
|
||||
/// engine. Null ⇒ the compiled defaults.</param>
|
||||
public DriverHostActor(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? 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<CommonsNodeId, bool>? 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
|
||||
/// <see cref="DependencyMuxTagUpstreamSource"/> + the wired <see cref="IAlarmStateStore"/> (the
|
||||
/// replicated <c>LocalDbAlarmConditionStateStore</c> 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 <c>Props</c>, so EACH
|
||||
/// incarnation (a supervision restart included) builds, owns and disposes its own engine; it stops
|
||||
/// with the driver host.
|
||||
/// </summary>
|
||||
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<ApplyWedged>(HandleApplyWedged);
|
||||
Receive<GetDiagnostics>(HandleGetDiagnostics);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
|
||||
Receive<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<ScriptedAlarmHostActor.ScriptedAlarmHostRestarted>(_ => HandleScriptedAlarmHostRestarted());
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
|
||||
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
|
||||
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
|
||||
@@ -1562,6 +1596,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
Receive<ApplyWedged>(HandleApplyWedged);
|
||||
Receive<GetDiagnostics>(HandleGetDiagnostics);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
|
||||
Receive<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<ScriptedAlarmHostActor.ScriptedAlarmHostRestarted>(_ => HandleScriptedAlarmHostRestarted());
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
|
||||
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
|
||||
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
|
||||
@@ -1606,6 +1642,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// </summary>
|
||||
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<DriverInstanceActor.AttributeValueBatchPublished>(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<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<ScriptedAlarmHostActor.ScriptedAlarmHostRestarted>(_ => HandleScriptedAlarmHostRestarted());
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// (<see cref="StopChild"/>) and an operator Restart (a new driver object). Deliberately NOT sent on a
|
||||
/// supervision restart (<see cref="HandleReprovisionRequested"/>): 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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 <see cref="RetireChild"/>s it, and <see cref="ForwardBatchToMux"/> drops every burst from a
|
||||
/// retired child.
|
||||
/// </remarks>
|
||||
/// <param name="driverInstanceId">The driver instance that ended.</param>
|
||||
private void EvictRetainedValues(string driverInstanceId) =>
|
||||
_dependencyMux?.Tell(new DependencyMuxActor.SourceRemoved(driverInstanceId));
|
||||
|
||||
/// <summary>Marks a child this host is about to stop as retired and watches it until it terminates.</summary>
|
||||
/// <param name="child">The child being stopped.</param>
|
||||
private void RetireChild(IActorRef child)
|
||||
{
|
||||
_retiredChildren.Add(child);
|
||||
Context.Watch(child);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>A retired child has terminated; nothing more can arrive from it.</summary>
|
||||
/// <param name="msg">The termination notice.</param>
|
||||
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);
|
||||
|
||||
|
||||
@@ -108,21 +108,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// state — a no-op when already Reconnecting or Connecting.
|
||||
/// </summary>
|
||||
public sealed record ForceReconnect;
|
||||
/// <summary>One driver-published value. <b>No production sender remains</b> — the subscription path
|
||||
/// batches every data change into an <see cref="AttributeValueBatchPublished"/>, and the comms-loss fan
|
||||
/// (<c>DriverHostActor.FanCommsLossToDependencyMux</c>), 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.
|
||||
/// <summary>One value within an <see cref="AttributeValueBatchPublished"/>.
|
||||
/// <para><paramref name="StatusCode"/> is the driver's full OPC UA status code including its sub-code
|
||||
/// — not a severity bucket. Classify it with <see cref="OpcUaStatus.IsGood"/> and friends rather
|
||||
/// than comparing it to <see cref="OpcUaStatus.Good"/>.</para></summary>
|
||||
public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
|
||||
/// <summary>One value within an <see cref="AttributeValueBatchPublished"/>.
|
||||
/// than comparing it to <see cref="OpcUaStatus.Good"/>.</para>
|
||||
/// <para>A struct deliberately: an <c>AttributeValueItem[]</c> 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
|
||||
/// <see cref="IReadOnlyList{T}"/> does not box.</para></summary>
|
||||
public readonly record struct AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
|
||||
/// <summary>Batched form of <see cref="AttributeValuePublished"/> — every changed attribute of ONE
|
||||
/// driver production burst in one message. The parent fans the whole batch to the mux and the OPC UA
|
||||
/// <summary>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.</summary>
|
||||
public sealed record AttributeValueBatchPublished(string DriverInstanceId, IReadOnlyList<AttributeValueItem> Items);
|
||||
private sealed record DataChangeForward(string FullReference, DataValueSnapshot Snapshot);
|
||||
@@ -132,7 +127,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
private sealed record DataChangeBatchForward(IReadOnlyList<DataChangeItem> Items);
|
||||
/// <summary>Published to the parent whenever the subscribed driver (an <see cref="IAlarmSource"/>) fires
|
||||
/// <see cref="IAlarmSource.OnAlarmEvent"/>. The parent (<see cref="DriverHostActor"/>) projects + routes it
|
||||
/// to the materialised Part 9 condition. Parallels <see cref="AttributeValuePublished"/>.</summary>
|
||||
/// to the materialised Part 9 condition. Parallels <see cref="AttributeValueBatchPublished"/>.</summary>
|
||||
public sealed record AttributeAlarmPublished(string DriverInstanceId, AlarmEventArgs Args);
|
||||
private sealed record NativeAlarmRaised(AlarmEventArgs Args);
|
||||
/// <summary>Self-sent on Connected entry / when alarm refs are (re)pushed, to establish the native-alarm
|
||||
|
||||
@@ -86,6 +86,21 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
/// <param name="Plans">The desired Equipment-namespace scripted-alarm plans.</param>
|
||||
public sealed record ApplyScriptedAlarms(IReadOnlyList<EquipmentScriptedAlarmPlan> Plans);
|
||||
|
||||
/// <summary>
|
||||
/// Told to the parent from <see cref="PostRestart"/>. 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 <see cref="ApplyScriptedAlarms"/> its inputs stop and every operator command is dropped as not
|
||||
/// owned. The parent (<c>DriverHostActor</c>) answers by re-sending the last plans it applied, mirroring
|
||||
/// how a driver child's <c>ReprovisionRequested</c> is answered.
|
||||
/// </summary>
|
||||
public sealed class ScriptedAlarmHostRestarted
|
||||
{
|
||||
/// <summary>The single instance.</summary>
|
||||
public static readonly ScriptedAlarmHostRestarted Instance = new();
|
||||
|
||||
private ScriptedAlarmHostRestarted() { }
|
||||
}
|
||||
|
||||
/// <summary>Marshals an engine emission off the engine's worker thread onto the actor thread.
|
||||
/// Carries the <see cref="ScriptedAlarmEvent"/> the engine raised on <c>OnEvent</c>.</summary>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>Tells the parent <see cref="ScriptedAlarmHostRestarted"/> so it re-sends the last
|
||||
/// <see cref="ApplyScriptedAlarms"/>: the new incarnation has nothing loaded and no mux interest.</remarks>
|
||||
protected override void PostRestart(Exception reason)
|
||||
{
|
||||
base.PostRestart(reason);
|
||||
Context.Parent.Tell(ScriptedAlarmHostRestarted.Instance);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
|
||||
@@ -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<IConfiguration>(), roleInfo, replicationPeerHost),
|
||||
// The HistoryUpdate verdict (closed until the first redundancy snapshot).
|
||||
historyUpdateRoleView: resolver.GetService<IHistoryUpdateRoleView>()),
|
||||
historyUpdateRoleView: resolver.GetService<IHistoryUpdateRoleView>(),
|
||||
// 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<ScriptingOptions>()),
|
||||
DriverHostActorName);
|
||||
registry.Register<DriverHostActorKey>(driverHost);
|
||||
|
||||
|
||||
@@ -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
|
||||
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> — one envelope per driver
|
||||
/// production burst, which the mux filters per item. The per-value
|
||||
/// <see cref="DriverInstanceActor.AttributeValuePublished"/> receive has <b>no production sender left
|
||||
/// at all</b>: the comms-loss fan (<c>DriverHostActor.FanCommsLossToDependencyMux</c>) was its last
|
||||
/// one and now fabricates an <c>AttributeValueBatchPublished</c> 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
|
||||
/// (<c>DriverHostActor.FanCommsLossToDependencyMux</c>) fabricates one too. There is no per-value
|
||||
/// ingress: a single value is a one-item batch (tests build one with <c>ValuePublishes.Single</c>).
|
||||
///
|
||||
/// <b>The mux emits ONLY batches</b>, 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;
|
||||
/// <b>Source removal.</b> The mux records which driver instance last published each retained ref;
|
||||
/// <see cref="SourceRemoved"/> 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 <see cref="SourceRemoved"/> yet —
|
||||
/// <c>DriverHostActor</c> (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. <c>DriverHostActor</c> sends <see cref="SourceRemoved"/> 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.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class DependencyMuxActor : ReceiveActor, IWithTimers
|
||||
@@ -154,7 +150,6 @@ public sealed class DependencyMuxActor : ReceiveActor, IWithTimers
|
||||
Receive<UnregisterInterest>(msg => RemoveSubscriber(msg.Subscriber));
|
||||
Receive<SourceRemoved>(OnSourceRemoved);
|
||||
Receive<SweepReleased>(_ => OnSweep());
|
||||
Receive<DriverInstanceActor.AttributeValuePublished>(OnAttributeValuePublished);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(OnAttributeValueBatchPublished);
|
||||
Receive<Terminated>(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.
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// <c>ctx.CancellationToken</c> — the run token, visible to scripts. <see cref="ScriptRunGuard"/> already
|
||||
/// cancels the token it hands <see cref="ScriptEvaluator{TContext, TResult}.RunAsync"/> 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.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class ScriptCancellationTokenTests
|
||||
{
|
||||
/// <summary>Spins until cancelled, but ends by itself after 10 s so a broken implementation cannot pin a
|
||||
/// pool thread past the test.</summary>
|
||||
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<FakeScriptContext, bool>.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<FakeScriptContext, int>.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<FakeScriptContext, int>.Compile(CooperativeSpin);
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
|
||||
await Should.ThrowAsync<ScriptTimeoutException>(
|
||||
() => 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<FakeScriptContext, int>.Compile("return 5;");
|
||||
(await guard.RunAsync("k", CooperativeSpin, fast, new FakeScriptContext(), TimeSpan.FromSeconds(5), ct))
|
||||
.ShouldBe(5);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="ScriptingOptions.Sanitized"/> — the bind-time clamp of the <c>Scripting</c> section's numeric
|
||||
/// knobs. An absurd value must be pulled into a sane range at bind time rather than surfacing later as an
|
||||
/// <see cref="OverflowException"/> inside a script run (<c>timeout * StuckOrphanQuarantineFactor</c>).
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <see cref="OverflowException"/>.
|
||||
/// </summary>
|
||||
[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<FakeScriptContext, int>.Compile(spin);
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var timeout = TimeSpan.FromMilliseconds(60);
|
||||
|
||||
await Should.ThrowAsync<ScriptTimeoutException>(
|
||||
() => guard.RunAsync("k", spin, eval, new FakeScriptContext(), timeout, ct));
|
||||
// The refusal path evaluates timeout * factor. Unclamped, TimeSpan * double.MaxValue throws OverflowException.
|
||||
await Should.ThrowAsync<ScriptRunRefusedException>(
|
||||
() => guard.RunAsync("k", spin, eval, new FakeScriptContext(), timeout, ct));
|
||||
// A huge timeout times the clamped maximum factor must not overflow either.
|
||||
await Should.ThrowAsync<ScriptRunRefusedException>(
|
||||
() => guard.RunAsync("k", spin, eval, new FakeScriptContext(), TimeSpan.FromDays(365 * 1000), ct));
|
||||
}
|
||||
}
|
||||
+69
@@ -0,0 +1,69 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// The Calculation driver's run-guard options: the node-wide <c>Scripting</c> 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.
|
||||
/// </summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>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.</summary>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
+2
-1
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
+64
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Review m-1: the <c>Scripting</c> 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.
|
||||
/// </summary>
|
||||
public sealed class ScriptingOptionsBindingTests
|
||||
{
|
||||
private static IConfiguration Config(Dictionary<string, string?> 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);
|
||||
}
|
||||
}
|
||||
+103
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Review I-2: under the factory Props a supervision restart of the scripted-alarm host builds a fresh, EMPTY
|
||||
/// engine. Nothing re-sent <c>ApplyScriptedAlarms</c>, 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 <see cref="ScriptedAlarms.ScriptedAlarmHostActor.ScriptedAlarmHostRestarted"/> from <c>PostRestart</c>,
|
||||
/// and <see cref="DriverHostActor"/> re-sends the last plans it applied. This test never re-applies by hand.
|
||||
/// </summary>
|
||||
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<ZB.MOM.WW.OtOpcUa.Configuration.OtOpcUaConfigDbContext> 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<string> { "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<DependencyMuxActor.RegisterInterest>(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<DependencyMuxActor.UnregisterInterest>(u => u.Subscriber.Equals(alarmHost), Timeout);
|
||||
|
||||
// The parent re-applies on the restart signal: interest is registered again by the new incarnation.
|
||||
mux.FishForMessage<DependencyMuxActor.RegisterInterest>(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<OpcUaPublishActor.AlarmStateUpdate>(
|
||||
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<OpcUaPublishActor.AlarmStateUpdate>(
|
||||
m => m.AlarmNodeId == "alm-1" && m.State.Acknowledged, Timeout)
|
||||
.State.Acknowledged.ShouldBeTrue();
|
||||
}
|
||||
}
|
||||
+224
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// The dependency mux retains a ref's last value for replay-opting subscribers and evicts a driver's values on
|
||||
/// <see cref="DependencyMuxActor.SourceRemoved"/>. 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.
|
||||
/// </summary>
|
||||
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<OtOpcUaConfigDbContext> _db = null!;
|
||||
|
||||
/// <summary>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.</summary>
|
||||
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<string> { "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<string>(StringComparer.Ordinal) { Path1, Path2 };
|
||||
_holder.FishForMessage<VirtualTagActor.DependencyValueBatch>(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);
|
||||
}
|
||||
|
||||
/// <summary>Waits until the holder has been delivered <paramref name="value"/> for <paramref name="path"/>.</summary>
|
||||
private void AwaitHolder(string path, object value) =>
|
||||
_holder.FishForMessage<VirtualTagActor.DependencyValueBatch>(
|
||||
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);
|
||||
|
||||
/// <summary>A fresh replay-opting subscriber for both refs; returns what it was replayed, keyed by ref.</summary>
|
||||
private Dictionary<string, VirtualTagActor.DependencyValueChanged> Replayed()
|
||||
{
|
||||
var probe = CreateTestProbe();
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, probe.Ref, ReplayRetained: true));
|
||||
return probe.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget)
|
||||
.Items.ToDictionary(i => i.TagId, StringComparer.Ordinal);
|
||||
}
|
||||
|
||||
/// <summary>The live child actor for <paramref name="driverId"/> (named <c>drv-{id}-g{generation}</c>).</summary>
|
||||
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<ActorIdentity>(PresenceBudget).Subject.ShouldNotBeNull();
|
||||
}
|
||||
|
||||
private static void ShouldNotBeTheOldGood(Dictionary<string, VirtualTagActor.DependencyValueChanged> 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);
|
||||
}
|
||||
|
||||
/// <summary>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.</summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>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).</summary>
|
||||
[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");
|
||||
}
|
||||
|
||||
/// <summary>Review m-3: an operator Restart builds a new driver object; the old one's value is evicted and the
|
||||
/// new one's replays.</summary>
|
||||
[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");
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -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));
|
||||
}
|
||||
|
||||
|
||||
@@ -279,11 +279,18 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla
|
||||
/// <inheritdoc />
|
||||
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <summary>
|
||||
/// When set, <see cref="ShutdownAsync"/> 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.
|
||||
/// </summary>
|
||||
public TaskCompletionSource? ShutdownGate { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Interlocked.Increment(ref _shutdownCount);
|
||||
return Task.CompletedTask;
|
||||
return ShutdownGate?.Task ?? Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -5,8 +5,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
/// <summary>
|
||||
/// Builds the ONE message shape a driver child sends for its values — a
|
||||
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> — for tests that push a single value into the
|
||||
/// dependency mux by hand. The per-value <see cref="DriverInstanceActor.AttributeValuePublished"/> 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).
|
||||
/// </summary>
|
||||
public static class ValuePublishes
|
||||
{
|
||||
|
||||
+1
-1
@@ -19,7 +19,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
|
||||
/// <remarks>
|
||||
/// Adapted from the plan's Task 17: the recorder handles the REAL fan-out message the mux emits —
|
||||
/// <see cref="VirtualTagActor.DependencyValueChanged"/> (TagId/Value/TimestampUtc/StatusCode) —
|
||||
/// not <c>DriverInstanceActor.AttributeValuePublished</c>. The typed-value / real-status behaviour is
|
||||
/// not the driver-side <c>DriverInstanceActor.AttributeValueBatchPublished</c>. The typed-value / real-status behaviour is
|
||||
/// pinned in <see cref="ContinuousHistorizationRecorderTypedValueTests"/>.
|
||||
/// </remarks>
|
||||
public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
|
||||
+14
-14
@@ -10,9 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
|
||||
|
||||
public sealed class DependencyMuxActorTests : RuntimeActorTestBase
|
||||
{
|
||||
/// <summary>Verifies that AttributeValuePublished messages are forwarded only to subscribers interested in the specific reference.</summary>
|
||||
/// <summary>Verifies that a published value is forwarded only to subscribers interested in the specific reference.</summary>
|
||||
[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<VirtualTagActor.DependencyValueBatch>();
|
||||
|
||||
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<VirtualTagActor.DependencyValueBatch>().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<VirtualTagActor.EvaluationResult>(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<VirtualTagActor.EvaluationResult>(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<VirtualTagActor.DependencyValueBatch>()
|
||||
|
||||
Reference in New Issue
Block a user