Merge local master 5135a036 into remed/lane-AbLegacy

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:36:11 -04:00
66 changed files with 4024 additions and 469 deletions
@@ -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"/>.
@@ -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,
};
}
}
@@ -117,6 +117,12 @@ public sealed class AbCipDriverOptions
/// device round trips (the whole point: a strictly sequential batch pays one full round
/// trip per tag); lowering it to <c>1</c> restores strictly sequential reads. Values below
/// <c>1</c> are clamped to <c>1</c> at the use site.
/// <para>
/// On the <see cref="AbCipTransport.Native"/> transport the effective window is at least 64: there the
/// window caps how many services one Multiple Service Packet coalesces (the device's shared scheduler packs
/// what is queued, bounded by the connection's request/reply byte budget), so a small value only multiplies
/// round trips. A value above 64 is honoured; a value below it is raised.
/// </para>
/// </summary>
public int MaxConcurrentReads { get; init; } = 8;
@@ -0,0 +1,80 @@
using System.Buffers.Binary;
using System.Collections;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// <summary>
/// Encodes a whole-array write (review 07 U10): <c>count</c> elements at the element type's wire width,
/// little-endian, exactly the layout a Logix atomic array holds and the read path decodes. Shared by both
/// runtimes so the bytes on the wire are transport-independent.
/// </summary>
/// <remarks>
/// <para><b>Element values CONVERT, they are not exact-matched.</b> ScadaBridge (and any OPC UA client that
/// round-trips what it read) writes CLR lists such as <c>int[]</c> to an INT array — the read path itself
/// publishes INT/SINT elements as <c>int</c> — so each element goes through the same <see cref="Convert"/> the
/// scalar path uses. A value that does not fit is an <see cref="OverflowException"/> (BadOutOfRange) and one that
/// cannot convert a <see cref="FormatException"/> / <see cref="InvalidCastException"/> (BadTypeMismatch), all
/// thrown before any byte is sent.</para>
/// <para>BOOL arrays (bit-packed DWORDs — a byte-per-element encode would clobber neighbour bits), STRING arrays
/// (structures) and structure arrays are not encodable here; the driver refuses them <c>BadNotSupported</c>
/// before a runtime exists.</para>
/// </remarks>
internal static class AbCipArrayCodec
{
/// <summary>Wire width of one element of an encodable array type, or <c>null</c> when the type is not encodable.</summary>
/// <param name="type">The element type.</param>
/// <returns>The element width in bytes, or <c>null</c>.</returns>
public static int? ElementWidth(AbCipDataType type) => type switch
{
AbCipDataType.SInt or AbCipDataType.USInt => 1,
AbCipDataType.Int or AbCipDataType.UInt => 2,
AbCipDataType.DInt or AbCipDataType.UDInt or AbCipDataType.Real or AbCipDataType.Dt => 4,
AbCipDataType.LInt or AbCipDataType.ULInt or AbCipDataType.LReal => 8,
_ => null,
};
/// <summary>
/// The elements of an array write value, or <c>null</c> when the value is not a list (a scalar, a string, null).
/// Any non-string <see cref="IEnumerable"/> is accepted — <c>int[]</c>, <c>short[]</c>, <c>List&lt;object&gt;</c> …
/// </summary>
/// <param name="value">The client's write value.</param>
/// <returns>The elements, or <c>null</c>.</returns>
public static object?[]? Elements(object? value) => value switch
{
null or string => null,
IEnumerable list => [.. list.Cast<object?>()],
_ => null,
};
/// <summary>Encodes <paramref name="values"/> as <paramref name="type"/> elements.</summary>
/// <param name="type">The element type (must have an <see cref="ElementWidth"/>).</param>
/// <param name="values">The element values; their count is the element count.</param>
/// <returns>The little-endian wire bytes, <c>values.Count × ElementWidth(type)</c> long.</returns>
/// <exception cref="NotSupportedException">The type is not encodable.</exception>
public static byte[] Encode(AbCipDataType type, IReadOnlyList<object?> values)
{
var width = ElementWidth(type)
?? throw new NotSupportedException($"Array writes of {type} elements are not supported.");
var buffer = new byte[values.Count * width];
for (var i = 0; i < values.Count; i++)
{
var v = values[i];
var span = buffer.AsSpan(i * width, width);
switch (type)
{
case AbCipDataType.SInt: span[0] = unchecked((byte)Convert.ToSByte(v)); break;
case AbCipDataType.USInt: span[0] = Convert.ToByte(v); break;
case AbCipDataType.Int: BinaryPrimitives.WriteInt16LittleEndian(span, Convert.ToInt16(v)); break;
case AbCipDataType.UInt: BinaryPrimitives.WriteUInt16LittleEndian(span, Convert.ToUInt16(v)); break;
case AbCipDataType.DInt:
case AbCipDataType.Dt: BinaryPrimitives.WriteInt32LittleEndian(span, Convert.ToInt32(v)); break;
case AbCipDataType.UDInt: BinaryPrimitives.WriteUInt32LittleEndian(span, Convert.ToUInt32(v)); break;
case AbCipDataType.LInt: BinaryPrimitives.WriteInt64LittleEndian(span, Convert.ToInt64(v)); break;
case AbCipDataType.ULInt: BinaryPrimitives.WriteUInt64LittleEndian(span, Convert.ToUInt64(v)); break;
case AbCipDataType.Real: BinaryPrimitives.WriteSingleLittleEndian(span, Convert.ToSingle(v)); break;
case AbCipDataType.LReal: BinaryPrimitives.WriteDoubleLittleEndian(span, Convert.ToDouble(v)); break;
}
}
return buffer;
}
}
File diff suppressed because it is too large Load Diff
@@ -18,10 +18,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// non-CIP server) returns <c>Ok = false</c> with a "handshake failed" message instead of
/// a false-positive green tick.
/// <para>
/// Status-code discrimination: <c>AbStatus.Ok</c> and <c>AbStatus.NotFound</c> mean the controller
/// answered the CIP session (a non-existent probe tag name still proves CIP connectivity) and are
/// treated as "reachable - Ok = true". Every other status (timeout, bad connection, bad reply,
/// partial, abort, ...) is a transport or session failure: "handshake failed - Ok = false".
/// Status-code discrimination: <c>AbStatus.Ok</c> and every device refusal (not found, out of bounds,
/// not allowed, too large/small, type mismatch, unsupported) mean the controller answered the CIP session
/// (a non-existent probe tag name still proves CIP connectivity) and are treated as "reachable - Ok = true".
/// Every other status (timeout, bad connection, bad reply, partial, abort, ...) is a transport or session
/// failure: "handshake failed - Ok = false".
/// </para>
/// </summary>
public sealed class AbCipDriverProbe : IDriverProbe
@@ -147,7 +148,7 @@ public sealed class AbCipDriverProbe : IDriverProbe
{
var msg = statusCode == AbStatus.Ok
? "CIP session OK"
: "CIP session OK (controller reachable; probe tag not found)";
: RefusedProbeMessage(statusCode);
return new(true, msg, sw.Elapsed);
}
@@ -164,7 +165,7 @@ public sealed class AbCipDriverProbe : IDriverProbe
if (IsReachableException(ex))
{
sw.Stop();
return new(true, "CIP session OK (controller reachable; probe tag not found)", sw.Elapsed);
return new(true, RefusedProbeMessage(ex.Status), sw.Elapsed);
}
return new(false, $"Reachable at {host}:{port} but CIP handshake failed: {ex.Message}", null);
@@ -180,13 +181,20 @@ public sealed class AbCipDriverProbe : IDriverProbe
}
/// <summary>
/// Returns <c>true</c> for transport statuses that indicate the CIP controller answered
/// but the probe tag name was not found or did not match — sufficient to confirm CIP
/// reachability. Returns <c>true</c> for <c>Ok</c> (tag was found).
/// Returns <c>true</c> for transport statuses that indicate the CIP controller answered — <c>Ok</c>, or any
/// device refusal (<see cref="AbCipRefusals.IsDeviceRefusal(AbStatus)"/>: not found, out of bounds, not
/// allowed, too large/small, type mismatch, unsupported) — sufficient to confirm CIP reachability.
/// </summary>
/// <param name="s">The transport status to classify.</param>
/// <returns><see langword="true"/> if the status confirms CIP reachability.</returns>
internal static bool IsReachableStatus(AbStatus s) => s is AbStatus.Ok or AbStatus.NotFound;
internal static bool IsReachableStatus(AbStatus s) => s == AbStatus.Ok || AbCipRefusals.IsDeviceRefusal(s);
/// <summary>The Test-Connect message for a controller that answered but refused the probe tag.</summary>
/// <param name="status">The refusal.</param>
/// <returns>The message.</returns>
private static string RefusedProbeMessage(AbStatus status) => status == AbStatus.NotFound
? "CIP session OK (controller reachable; probe tag not found)"
: $"CIP session OK (controller reachable; probe tag refused: {status})";
/// <summary>
/// Decides whether an <see cref="AbTransportException"/> is a "controller answered CIP but the
@@ -198,7 +206,8 @@ public sealed class AbCipDriverProbe : IDriverProbe
internal static bool IsReachableException(AbTransportException ex)
{
ArgumentNullException.ThrowIfNull(ex);
return IsReachableStatus(ex.Status);
// A local refusal (the library / driver refused without any I/O) proves nothing about the controller.
return ex is not AbCipLocalRefusalException && IsReachableStatus(ex.Status);
}
private static (AbCipHostAddress? parsed, AbCipDeviceOptions? device) ExtractTarget(AbCipDriverOptions opts)
@@ -0,0 +1,18 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// <summary>
/// Thrown by <see cref="AbCipDriver.InitializeAsync"/> when a shutdown, dispose or newer initialization retired the
/// driver's tables while this initialization was building, so it published nothing. Deliberately NOT an
/// <see cref="OperationCanceledException"/>: nobody cancelled this call, and a caller that treats cancellation as
/// "my own token fired" must not mistake a superseded attempt for one. The superseding operation owns the driver's
/// state; the caller (normally the runtime, which has already abandoned this attempt) can ignore the result.
/// </summary>
/// <remarks>
/// Resilience accounting (#737 m2): not being an OCE, it passes the Initialize pipeline's breaker
/// <c>ShouldHandle</c> (<c>DriverResiliencePipelineBuilder</c>) and, if its generation still matched, the actor's
/// <c>RecordFault</c>. That is practically unreachable — <c>DriverInstanceActor</c> runs initialisation single-flight
/// and drops generation-mismatched results — and a superseded attempt is at worst one counted failure. Recognising
/// it there is a runtime-side (Lane A) change if it is ever wanted.
/// </remarks>
/// <param name="message">The detail.</param>
public sealed class AbCipInitializationSupersededException(string message) : InvalidOperationException(message);
@@ -0,0 +1,15 @@
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// <summary>
/// A refusal raised LOCALLY — by the driver's own runtime or by libplctag (<c>ErrorNotImplemented</c>) — before or
/// without any reply from the controller. It carries <see cref="AbStatus.Unsupported"/> (so it maps to
/// <c>BadNotSupported</c>) but, unlike a device refusal, proves nothing about the device: it keeps the runtime and
/// writes no health, does NOT advance <c>LastSuccessfulRead</c>, records nothing in the connect backoff, and is not
/// "reachable" for the probe or Test Connect (review 07 / Gitea #737 I2).
/// </summary>
/// <param name="message">The refusal detail.</param>
/// <param name="inner">The underlying exception, if any.</param>
public sealed class AbCipLocalRefusalException(string message, Exception? inner = null)
: AbTransportException(AbStatus.Unsupported, message, inner);
@@ -0,0 +1,42 @@
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// <summary>
/// The ONE classification of a transport status into "the controller answered" vs "the wire failed", shared by
/// the data path, the probe loop, the Test-Connect probe and the native runtime's initialisation.
/// </summary>
/// <remarks>
/// A device refusal is deterministic: the session works and the controller said no to this one tag (a renamed
/// tag, <c>Arr[999]</c>, a read-protected tag, a type the authored config does not match). It is per-tag Bad,
/// keeps the tag runtime (evicting re-creates it — and, for a sole lease holder, re-Forward-Opens — on every poll),
/// never writes <c>Degraded</c>, never opens the device-wide connect backoff, and warns once per tag. Everything
/// else (timeout, bad connection, a reply that failed to parse, partial, abort, …) is a comms fault.
/// <para>
/// The native transport currently folds CIP general statuses 0x1C, 0x20 and 0xFF/0x2104 into
/// <see cref="AbStatus.BadReply"/> (<c>AbStatusCodes.FromCip</c>, owned by <c>Driver.Ab.Transport</c>), which
/// this classification cannot tell from a garbled reply; until the transport maps them explicitly they stay
/// comms faults.
/// </para>
/// </remarks>
internal static class AbCipRefusals
{
/// <summary>True when <paramref name="status"/> proves the controller answered but refused this tag.</summary>
/// <param name="status">The transport status.</param>
/// <returns><see langword="true"/> for a deterministic device refusal.</returns>
public static bool IsDeviceRefusal(AbStatus status) => status is AbStatus.NotFound or AbStatus.OutOfBounds
or AbStatus.NotAllowed or AbStatus.TooLarge or AbStatus.TooSmall or AbStatus.TypeMismatch
or AbStatus.Unsupported;
/// <summary>True when <paramref name="status"/> (an <see cref="AbStatus"/> value) is a device refusal.</summary>
/// <param name="status">The runtime status word.</param>
/// <returns><see langword="true"/> for a deterministic device refusal.</returns>
public static bool IsDeviceRefusal(int status) => IsDeviceRefusal((AbStatus)status);
/// <summary>True when <paramref name="ex"/> is an <see cref="AbTransportException"/> carrying a device refusal.</summary>
/// <param name="ex">The exception.</param>
/// <returns><see langword="true"/> for a deterministic device refusal.</returns>
/// <remarks>An <see cref="AbCipLocalRefusalException"/> is never a device refusal: nothing reached the device.</remarks>
public static bool IsDeviceRefusal(Exception ex) =>
ex is AbTransportException ate and not AbCipLocalRefusalException && IsDeviceRefusal(ate.Status);
}
@@ -15,6 +15,7 @@ public static class AbCipStatusMapper
public const uint GoodMoreData = 0x00A60000u;
public const uint BadInternalError = 0x80020000u;
public const uint BadNodeIdUnknown = 0x80340000u;
public const uint BadNotReadable = 0x803A0000u;
public const uint BadNotWritable = 0x803B0000u;
public const uint BadOutOfRange = 0x803C0000u;
public const uint BadNotSupported = 0x803D0000u;
@@ -51,4 +52,14 @@ public static class AbCipStatusMapper
/// <returns>The OPC UA status code.</returns>
public static uint MapException(Exception ex) =>
AbOpcUaStatusMap.MapException(ex);
/// <summary>
/// Status for a READ the controller refused: as <see cref="MapStatus(AbStatus)"/>, except that a
/// <see cref="AbStatus.NotAllowed"/> answer to a read is <see cref="BadNotReadable"/> (the shared table's
/// <c>BadNotWritable</c> describes a refused write).
/// </summary>
/// <param name="status">The refusal status.</param>
/// <returns>The OPC UA status code.</returns>
public static uint MapReadStatus(AbStatus status) =>
status == AbStatus.NotAllowed ? BadNotReadable : MapStatus(status);
}
@@ -25,6 +25,30 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// </remarks>
public static class AbCipUdtMemberLayout
{
/// <summary>Layouts already computed, keyed by the parent definition's member list (reference identity).</summary>
private static readonly System.Runtime.CompilerServices.ConditionalWeakTable<IReadOnlyList<AbCipStructureMember>, LayoutBox> Cache = new();
/// <summary>Holds a computed layout, including a "not groupable" (<c>null</c>) verdict.</summary>
private sealed class LayoutBox(IReadOnlyDictionary<string, int>? offsets)
{
public IReadOnlyDictionary<string, int>? Offsets { get; } = offsets;
}
/// <summary>
/// <see cref="TryBuild"/>, computed once per member list (review 07 P11: the read planner rebuilt the layout —
/// a dictionary allocation and an O(members) walk — for every member reference on every poll, O(members²) per
/// parent per tick). A parent definition's member list is immutable for the life of the tag table that holds
/// it, so caching by list identity is exact; a re-init builds new lists and the old entries are collected with
/// them (<see cref="System.Runtime.CompilerServices.ConditionalWeakTable{TKey,TValue}"/>).
/// </summary>
/// <param name="members">The parent definition's declared members.</param>
/// <returns>The member offsets, or <c>null</c> when the members cannot be laid out by declaration.</returns>
public static IReadOnlyDictionary<string, int>? GetOrBuild(IReadOnlyList<AbCipStructureMember> members)
{
ArgumentNullException.ThrowIfNull(members);
return Cache.GetValue(members, m => new LayoutBox(TryBuild(m))).Offsets;
}
/// <summary>
/// Try to compute member offsets for the supplied declared members. Returns <c>null</c>
/// if any member type is unsupported for declaration-only layout.
@@ -67,7 +67,7 @@ public static class AbCipUdtReadPlanner
continue;
}
var offsets = AbCipUdtMemberLayout.TryBuild(parent.Members);
var offsets = AbCipUdtMemberLayout.GetOrBuild(parent.Members);
if (offsets is null || !offsets.TryGetValue(memberName, out var offset))
{
fallback.Add(new AbCipUdtReadFallback(i, name));
@@ -85,6 +85,17 @@ public interface IAbCipTagRuntime : IDisposable
/// <param name="bitIndex">Bit index for BOOL-within-DINT insertion, or null.</param>
/// <param name="value">Value to encode.</param>
void EncodeValue(AbCipDataType type, int? bitIndex, object? value);
/// <summary>
/// Encode a WHOLE-array write of <paramref name="values"/> (<c>values.Count</c> elements of
/// <paramref name="type"/>, at the element width — see <see cref="AbCipArrayCodec"/>) into the local buffer.
/// Callers pair this with <see cref="WriteAsync"/>. A runtime that cannot write arrays throws
/// <see cref="NotSupportedException"/>.
/// </summary>
/// <param name="type">The array element type.</param>
/// <param name="values">The element values; already length-checked against the tag's element count.</param>
void EncodeArray(AbCipDataType type, IReadOnlyList<object?> values) =>
throw new NotSupportedException($"{GetType().Name} does not support array writes.");
}
/// <summary>
@@ -14,11 +14,21 @@ internal static class LibplctagStatusTranslation
Status.Ok => AbStatus.Ok,
Status.Pending => AbStatus.Pending,
Status.ErrorTimeout => AbStatus.Timeout,
Status.ErrorNotFound or Status.ErrorNoMatch or Status.ErrorBadDevice => AbStatus.NotFound,
// #737 I2: only ErrorNotFound is the controller saying "no such tag". ErrorNoMatch / ErrorBadDevice describe a
// session or device-level failure (no proof the controller answered this tag), so they are comms-class: with
// the widened refusal classification, folding them into NotFound made a broken session read as "reachable".
Status.ErrorNotFound => AbStatus.NotFound,
Status.ErrorNoMatch or Status.ErrorBadDevice => AbStatus.BadConnection,
Status.ErrorNotAllowed => AbStatus.NotAllowed,
Status.ErrorOutOfBounds => AbStatus.OutOfBounds,
Status.ErrorTooLarge => AbStatus.TooLarge,
Status.ErrorTooSmall => AbStatus.TooSmall,
// ErrorNotImplemented is the LIBRARY refusing (no I/O); the exception paths raise it as an
// AbCipLocalRefusalException so it never counts as "the device answered" (see FromException).
// ASSUMPTION (#737 m3): libplctag reports a failed operation by THROWING, so a NotImplemented reaches the driver
// only through FromException. AbStatus has no "local" member (it is Driver.Ab.Transport's, Lane K), so a
// NotImplemented observed only as a STATUS WORD (GetStatus after a non-throwing op) would still classify as a
// device Unsupported refusal; that combination is not produced by libplctag.NET 1.5.2.
Status.ErrorUnsupported or Status.ErrorNotImplemented => AbStatus.Unsupported,
Status.ErrorPartial => AbStatus.Partial,
Status.ErrorAbort => AbStatus.Abort,
@@ -70,9 +80,21 @@ internal static class LibplctagStatusTranslation
public static async Task Wrap(Func<Task> op)
{
try { await op().ConfigureAwait(false); }
catch (LibPlcTagException ex) { throw new AbTransportException(ToAbStatus(ex), ex.Message, ex); }
catch (LibPlcTagException ex) { throw FromException(ex); }
}
/// <summary>
/// The transport exception for a libplctag failure: <c>ErrorNotImplemented</c> becomes an
/// <see cref="AbCipLocalRefusalException"/> (a library refusal, no I/O); everything else an
/// <see cref="AbTransportException"/> carrying the translated status.
/// </summary>
/// <param name="ex">The libplctag exception.</param>
/// <returns>The exception to throw.</returns>
public static AbTransportException FromException(LibPlcTagException ex) =>
Enum.TryParse<Status>((ex.Message ?? string.Empty).Trim(), ignoreCase: true, out var s) && s == Status.ErrorNotImplemented
? new AbCipLocalRefusalException($"libplctag does not implement this operation: {ex.Message}", ex)
: new AbTransportException(ToAbStatus(ex), ex.Message ?? string.Empty, ex);
/// <summary>Runs <paramref name="op"/>, rethrowing any <see cref="LibPlcTagException"/> as <see cref="AbTransportException"/>.</summary>
/// <typeparam name="T">The result type.</typeparam>
/// <param name="op">The libplctag operation.</param>
@@ -80,7 +102,7 @@ internal static class LibplctagStatusTranslation
public static async Task<T> Wrap<T>(Func<Task<T>> op)
{
try { return await op().ConfigureAwait(false); }
catch (LibPlcTagException ex) { throw new AbTransportException(ToAbStatus(ex), ex.Message, ex); }
catch (LibPlcTagException ex) { throw FromException(ex); }
}
/// <summary>Runs a synchronous libplctag call, rethrowing <see cref="LibPlcTagException"/> as <see cref="AbTransportException"/>.</summary>
@@ -90,7 +112,7 @@ internal static class LibplctagStatusTranslation
public static T WrapSync<T>(Func<T> op)
{
try { return op(); }
catch (LibPlcTagException ex) { throw new AbTransportException(ToAbStatus(ex), ex.Message, ex); }
catch (LibPlcTagException ex) { throw FromException(ex); }
}
/// <summary>Runs a synchronous libplctag call, rethrowing <see cref="LibPlcTagException"/> as <see cref="AbTransportException"/>.</summary>
@@ -98,6 +120,6 @@ internal static class LibplctagStatusTranslation
public static void WrapSync(Action op)
{
try { op(); }
catch (LibPlcTagException ex) { throw new AbTransportException(ToAbStatus(ex), ex.Message, ex); }
catch (LibPlcTagException ex) { throw FromException(ex); }
}
}
@@ -29,8 +29,19 @@ internal sealed class LibplctagTagRuntime : IAbCipTagRuntime
public LibplctagTagRuntime(AbCipTagCreateParams p)
{
_tag = BuildTag(p);
_boolArrayElement = !p.IsArray && p.AuthoredType == AbCipDataType.Bool
&& NativeAbCipTagRuntime.TrySplitBoolArrayElement(p.TagName, out _, out _);
}
/// <summary>
/// A scalar BOOL authored over a BOOL-array element (<c>Bits[n]</c>). libplctag reads it as whatever the
/// controller returns for that element segment — a packed DWORD — and exposes no type code, so which DWORD
/// came back (and so which bit is element n) cannot be checked here. Review 07 U11: such a read used to decode
/// "low byte != 0" as Good; it is refused (<c>BadNotSupported</c>) when a DWORD came back. The native transport
/// (the default) addresses the right DWORD and decodes the bit.
/// </summary>
private readonly bool _boolArrayElement;
/// <summary>Builds the (uninitialized) libplctag tag for <paramref name="p"/>.</summary>
/// <param name="p">The tag creation parameters.</param>
/// <returns>The new, uninitialised libplctag tag.</returns>
@@ -120,7 +131,10 @@ internal sealed class LibplctagTagRuntime : IAbCipTagRuntime
{
AbCipDataType.Bool => bitIndex is int bit
? _tag.GetBit(bit)
: _tag.GetInt8(offset) != 0,
: _boolArrayElement && offset == 0 && _tag.GetSize() == 4
? throw new AbTransportException(AbStatus.Unsupported,
$"'{_tag.Name}' is a BOOL-array element read as a packed DWORD; libplctag cannot say which DWORD it is — use the native transport.")
: _tag.GetInt8(offset) != 0,
AbCipDataType.SInt => (int)(sbyte)_tag.GetInt8(offset),
AbCipDataType.USInt => (int)_tag.GetUInt8(offset),
AbCipDataType.Int => (int)_tag.GetInt16(offset),
@@ -221,6 +235,23 @@ internal sealed class LibplctagTagRuntime : IAbCipTagRuntime
_ => 4,
};
/// <inheritdoc />
/// <remarks>
/// The codec's bytes are copied into the buffer the last read sized (<c>elem_count</c> elements), so the write
/// sends exactly the new array — never a stale tail. A buffer smaller than the encoded array is refused
/// (<see cref="AbStatus.TooSmall"/>) before anything is changed.
/// </remarks>
public void EncodeArray(AbCipDataType type, IReadOnlyList<object?> values) =>
LibplctagStatusTranslation.WrapSync(() =>
{
var bytes = AbCipArrayCodec.Encode(type, values);
var size = _tag.GetSize();
if (size < bytes.Length)
throw new AbTransportException(AbStatus.TooSmall,
$"Array write to '{_tag.Name}': {bytes.Length} byte(s) encoded but the tag buffer holds {size}.");
for (var i = 0; i < bytes.Length; i++) _tag.SetUInt8(i, bytes[i]);
});
/// <inheritdoc />
public void EncodeValue(AbCipDataType type, int? bitIndex, object? value) =>
LibplctagStatusTranslation.WrapSync(() => EncodeValueCore(type, bitIndex, value));
@@ -41,6 +41,34 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
private int? _implicitBit;
private byte[]? _path;
// U11: a scalar BOOL authored over a BOOL-array element "Bits[n]" — the bit (n % 32) of DWORD n / 32 when the
// array is packed. Null for every other tag.
private int? _boolElementBit;
// Both addressings of such a tag: "Bits[n / 32]" (packed: a Logix BOOL array is stored as 0xD3 DWORDs) and the
// original "Bits[n]" (per-element: a family that answers 0xC1). _path is whichever _boolLayout selects.
private byte[]? _dwordPath;
private byte[]? _elementPath;
/// <summary>How a BOOL-array element tag's array is laid out, as the controller's REPLY TYPE last showed it.</summary>
private enum BoolLayout
{
/// <summary>No reply type seen yet (or only refusals / comms faults): request DWORD-addressed, keep asking.</summary>
Unsettled,
/// <summary>The controller answered 0xD3: address DWORD n / 32, decode bit n % 32.</summary>
Packed,
/// <summary>The controller answered 0xC1: address element n, decode its byte.</summary>
PerElement,
}
// #737 fix round 2 (N1): ONLY a reply type settles (or re-settles) the layout — 0xD3 ⇒ Packed, 0xC1 ⇒ PerElement.
// A refusal or comms fault says nothing about the layout and leaves it as it was, so a tag deployed before the
// online edit that creates its array starts reading correctly once the array exists. A reply whose type disagrees
// with the addressing it was asked under is DISCARDED (never decoded) and re-read under the right addressing.
private BoolLayout _boolLayout;
private ConnectionLease? _lease;
private RequestScheduler? _privateScheduler; // ForTest only — production always uses the lease's
private LogixTagClient? _client;
@@ -51,6 +79,7 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
private AbStatus _status = AbStatus.Pending;
private byte[]? _writeBuffer;
private ushort _writeElements = 1;
private string? _pendingString;
private bool _disposed;
@@ -94,16 +123,15 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
/// <summary>The trailing bit selector split off the tag name (<c>Tag.5</c>), if any.</summary>
internal int? ImplicitBit => _implicitBit;
// Performs one read to prove the tag exists: libplctag's InitializeAsync errors for an unknown
// tag, and the driver's backoff classification depends on create throwing AbStatus.NotFound.
// Performs one read to prove the session works and learn the tag's type.
//
// A name the encoder refuses throws AbCipTagConfigurationException before any lease: it is this
// tag's configuration, not the device, and must not black out the device's other tags. A read the
// controller answers with a deterministic refusal (AbStatus.TypeMismatch / AbStatus.Unsupported)
// proves the tag exists, so it completes the initialization with that status rather than throwing:
// a throw here makes the driver dispose the runtime and re-create it - and, for a sole lease
// holder, re-Forward-Open - on every poll (mistype churn). The driver's own read reports the
// refusal.
// controller answers with a deterministic refusal (AbCipRefusals.IsDeviceRefusal: NotFound, OutOfBounds,
// NotAllowed, TooLarge/TooSmall, TypeMismatch, Unsupported) proves the session works, so it completes the
// initialization with that status rather than throwing: a throw here makes the driver dispose the runtime
// and re-create it - and, for a sole lease holder, re-Forward-Open - on every poll (a renamed tag or a
// mistype churned the lease). The driver's own read reports the refusal, per tag, on the kept runtime.
/// <inheritdoc />
public async Task InitializeAsync(CancellationToken cancellationToken)
{
@@ -116,7 +144,20 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
try
{
split = CipPath.SplitBitSelector(_p.TagName);
path = CipPath.EncodeSymbolic(split.Name);
var name = split.Name;
if (split.Bit is null && !_p.IsArray && _p.AuthoredType == AbCipDataType.Bool
&& TrySplitBoolArrayElement(name, out var dwordName, out var bit))
{
_elementPath = CipPath.EncodeSymbolic(name);
_dwordPath = CipPath.EncodeSymbolic(dwordName);
_boolElementBit = bit;
_boolLayout = BoolLayout.Unsettled;
path = _dwordPath;
}
else
{
path = CipPath.EncodeSymbolic(name);
}
}
catch (Exception ex) when (ex is ArgumentException or FormatException or OverflowException)
{
@@ -137,20 +178,45 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
{
await ReadAsync(cancellationToken).ConfigureAwait(false);
}
catch (AbTransportException ex) when (ex.Status is AbStatus.TypeMismatch or AbStatus.Unsupported)
catch (AbTransportException ex) when (AbCipRefusals.IsDeviceRefusal(ex))
{
// The controller answered — the tag exists. _status already holds the refusal.
// The controller answered. _status already holds the refusal.
}
}
/// <inheritdoc />
public async Task ReadAsync(CancellationToken cancellationToken)
public Task ReadAsync(CancellationToken cancellationToken) => ReadCoreAsync(cancellationToken, mayRelayout: true);
private async Task ReadCoreAsync(CancellationToken cancellationToken, bool mayRelayout)
{
var client = Client();
var path = Path();
try
{
var (type, handle, data) = await client.ReadAsync(path, _elements, cancellationToken, ExpectedElementBytes(_p.AuthoredType)).ConfigureAwait(false);
if (_boolElementBit is not null)
{
BoolLayout? shown = type == LogixTypeCode.BoolArray ? BoolLayout.Packed
: type == LogixTypeCode.Bool ? BoolLayout.PerElement
: null; // any other type (a mistyped tag): no layout evidence; the type check below reports it
if (shown is { } layout)
{
var askedPacked = ReferenceEquals(path, _dwordPath);
SetBoolLayout(layout);
if ((layout == BoolLayout.Packed) != askedPacked)
{
// The reply is for the OTHER addressing (element n / 32 of a per-element array, or DWORD n of a
// packed one) — not this element. Discard it and ask again under the layout it just showed.
if (mayRelayout)
{
await ReadCoreAsync(cancellationToken, mayRelayout: false).ConfigureAwait(false);
return;
}
throw new AbTransportException(AbStatus.TypeMismatch,
$"'{_p.TagName}': the controller's BOOL-array layout changed between two reads; retrying next read.");
}
}
}
_data = data;
_typeCode = type;
_structHandle = handle;
@@ -164,6 +230,13 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
}
}
/// <summary>Settles a BOOL-array element tag's layout and selects the matching request path.</summary>
private void SetBoolLayout(BoolLayout layout)
{
_boolLayout = layout;
_path = layout == BoolLayout.PerElement ? _elementPath : _dwordPath;
}
/// <summary>
/// Upper bound on one element's wire bytes for the scheduler's reply budget; 0 when the
/// authored type is unknown or a structure, which sends the read alone rather than guess.
@@ -196,6 +269,14 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
try
{
// An unsettled BOOL-array element layout is settled by a read first, so a write never goes to the
// DWORD-addressed element of an element-addressed array (#737 I1). A packed layout's write is refused below
// (0xD3 DWORD — it would clobber 31 neighbours); a per-element one writes "Bits[n]" as a BOOL.
if (_boolElementBit is not null && _boolLayout == BoolLayout.Unsettled)
{
await ReadAsync(cancellationToken).ConfigureAwait(false);
path = Path();
}
if (_pendingString is { } text)
{
// A STRING writes as the type the controller reported — a Logix STRING is a structure
@@ -204,7 +285,7 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
if (_typeCode is null || (_typeCode == LogixTypeCode.Struct && _structHandle is null))
await ReadAsync(cancellationToken).ConfigureAwait(false);
if (_typeCode == LogixTypeCode.Struct && _structHandle is null)
throw new AbTransportException(AbStatus.Unsupported,
throw new AbCipLocalRefusalException(
$"STRING write to '{_p.TagName}': the controller returned no structure handle.");
var payload = EncodeStringPayload(text);
await client.WriteAsync(path, _typeCode!.Value, _structHandle, 1, payload, cancellationToken)
@@ -213,7 +294,7 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
else if (_writeBuffer is { } buffer)
{
var type = _typeCode ?? AtomicCode(_p.AuthoredType)
?? throw new AbTransportException(AbStatus.Unsupported,
?? throw new AbCipLocalRefusalException(
$"Write to '{_p.TagName}': type unknown — no prior read and no authored atomic type.");
// The read rule decides first: a same-width mismatch (REAL under a DINT-authored tag)
// would otherwise send the int bit pattern as a REAL and report Good.
@@ -223,17 +304,18 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
// Width is checked only for codes with a known width: DATE / DATE_AND_TIME (0xCD/0xCF) and
// Micro800 BYTE/WORD/LWORD (0xD1/0xD2/0xD4) are outside LogixTypeCode's table and write as today.
var width = LogixTypeCode.ByteWidth(type);
if (width > 0 && width != buffer.Length)
if (width > 0 && width * _writeElements != buffer.Length)
{
// An authored BOOL over a packed BOOL-array DWORD: writing the DWORD would clobber the
// 31 sibling bools (libplctag does exactly that), so refuse it as a config error.
if (type == LogixTypeCode.BoolArray)
if (type == LogixTypeCode.BoolArray && _writeElements == 1)
throw new NotSupportedException(
$"Write to '{_p.TagName}': a BOOL-array element (type 0xD3 DWORD) cannot be written as a single BOOL.");
throw new AbTransportException(AbStatus.TypeMismatch,
$"Write to '{_p.TagName}': encoded {buffer.Length} byte(s) but the controller's type 0x{type:X4} is {width} byte(s) wide.");
$"Write to '{_p.TagName}': encoded {buffer.Length} byte(s) for {_writeElements} element(s) but the controller's type 0x{type:X4} is {width} byte(s) wide.");
}
await client.WriteAsync(path, type, null, 1, buffer, cancellationToken).ConfigureAwait(false);
// U10: an array write is ONE Write Tag of _writeElements elements (the client fragments it if large).
await client.WriteAsync(path, type, null, _writeElements, buffer, cancellationToken).ConfigureAwait(false);
}
else
{
@@ -276,7 +358,7 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
return type switch
{
AbCipDataType.Bool => bitIndex is int bit ? GetBit(bit) : Span(offset, 1)[0] != 0,
AbCipDataType.Bool => bitIndex is int bit ? GetBit(bit) : DecodeScalarBool(offset),
AbCipDataType.SInt => (int)(sbyte)Span(offset, 1)[0],
AbCipDataType.USInt => (int)Span(offset, 1)[0],
AbCipDataType.Int => (int)BinaryPrimitives.ReadInt16LittleEndian(Span(offset, 2)),
@@ -315,10 +397,22 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
};
}
/// <inheritdoc />
public void EncodeArray(AbCipDataType type, IReadOnlyList<object?> values)
{
ArgumentNullException.ThrowIfNull(values);
_writeBuffer = null;
_pendingString = null;
var buffer = AbCipArrayCodec.Encode(type, values); // throws (nothing staged) before any state changes
_writeElements = (ushort)Math.Clamp(values.Count, 1, ushort.MaxValue);
_writeBuffer = buffer;
}
/// <inheritdoc />
public void EncodeValue(AbCipDataType type, int? bitIndex, object? value)
{
_writeBuffer = null;
_writeElements = 1;
_pendingString = null;
switch (type)
{
@@ -410,6 +504,47 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime
catch (Exception) { /* ForTest only; Dispose must not throw */ }
}
/// <summary>
/// A scalar BOOL. A packed (0xD3) reply is decoded ONLY as the element's own bit of the DWORD the packed layout
/// addressed; any other 0xD3 reply (no element bit, or not settled packed) is refused rather than decoded as
/// "byte 0 != 0" — that value would be some other bit, published as Good (#737 N1).
/// </summary>
private bool DecodeScalarBool(int offset)
{
if (offset == 0 && _typeCode == LogixTypeCode.BoolArray)
{
if (_boolElementBit is int element && _boolLayout == BoolLayout.Packed) return GetBit(element);
throw new AbTransportException(AbStatus.TypeMismatch,
$"'{_p.TagName}' is a scalar BOOL but the controller answered a packed BOOL-array DWORD (0xD3) with no element bit to select.");
}
return Span(offset, 1)[0] != 0;
}
/// <summary>
/// Splits a BOOL-array element reference <c>Name[n]</c> (one non-negative subscript, last segment) into the
/// DWORD it lives in (<c>Name[n / 32]</c>) and its bit within that DWORD (<c>n % 32</c>). A Logix BOOL array
/// is stored as DWORDs (type 0xD3) and its element segment addresses those DWORDs.
/// </summary>
/// <param name="name">The tag name, bit selector already split off.</param>
/// <param name="dwordName">The name addressing the containing DWORD.</param>
/// <param name="bit">The element's bit within that DWORD.</param>
/// <returns><see langword="true"/> when <paramref name="name"/> ends in a single subscript.</returns>
internal static bool TrySplitBoolArrayElement(string name, out string dwordName, out int bit)
{
dwordName = name;
bit = 0;
if (!name.EndsWith(']')) return false;
var open = name.LastIndexOf('[');
if (open <= 0) return false;
var inner = name[(open + 1)..^1];
if (inner.Length == 0 || !inner.All(char.IsAsciiDigit)
|| !int.TryParse(inner, System.Globalization.NumberStyles.None, System.Globalization.CultureInfo.InvariantCulture, out var n))
return false;
dwordName = $"{name[..open]}[{n / 32}]";
bit = n % 32;
return true;
}
private byte[] Path() =>
_path ?? throw new InvalidOperationException("InitializeAsync has not been called.");
@@ -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
@@ -21,7 +21,7 @@
<div class="col-md-3">
<label class="form-label">Max concurrent reads</label>
<InputNumber @bind-Value="_form.MaxConcurrentReads" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
<div class="form-text">Read work items in flight at once. Default 8; 1 = sequential.</div>
<div class="form-text">Read work items in flight at once. Default 8; 1 = sequential (libplctag). The native transport always uses at least 64 — it packs concurrent reads into one request.</div>
</div>
<div class="col-md-3">
<div class="form-check form-switch mt-4">
@@ -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);
+13 -1
View File
@@ -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.