Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs
T
Joseph Doherty 43b81b6643 test(abcip): live tags keep a per-element value check; close barrier and GC/RTT attribution (T21 review r1 I1/M9/M8/M4)
I1: a live verdict no longer waives the value check. Only elements that moved between the two libplctag reads are excused.
- An element equal in both libplctag reads must also be equal through native; otherwise **NO: stable element i**.
- A moved element must not read NaN or infinity on native unless libplctag saw it too; otherwise **NO: non-finite element i**.
- Live tags log their full raw bytes and every element, and the summary reports live tags as N of the total.
- The offline pin adds the pass-1 Real[5] shape: a stable element shifted on native fails. It is red with the element check disabled.

M9: after each native phase (differential and latency), wait for the session close. With AB_DIFF_COUNT_OPENS=1, assert exactly one Forward Close was sent, using the new CipConnection counters; acked is logged.

M8: each latency pass logs its GC collection deltas (gen0/1/2) for both transports, plus native per-exchange round-trip min/mean/max.

M4: the latency test also asserts that no libplctag orphan had to be rooted.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-07 20:16:33 -04:00

689 lines
35 KiB
C#

using System.Collections;
using System.Diagnostics;
using System.Net;
using System.Net.NetworkInformation;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;
/// <summary>One tag the harness reads, with the type it decodes under (null ⇒ raw comparison only).</summary>
internal sealed record SelectedTag(string Name, AbCipDataType? Type, int ElementCount, bool IsArray, string Origin);
/// <summary>What one transport observed for one tag.</summary>
/// <param name="TypeCode">The CIP type code the reply carried, where the transport exposes it (native only: libplctag.NET
/// 1.5.2 does not surface it).</param>
internal sealed record ReadObservation(
int Status, string? Error, byte[]? Raw, object? Decoded, string? DecodeError = null, ushort? TypeCode = null)
{
public bool Succeeded => Status == (int)AbStatus.Ok && Error is null;
}
/// <summary>What one transport observed over its whole phase.</summary>
internal sealed class TransportObservations
{
public List<AbCipDiscoveredTag> Browse { get; } = [];
public string? BrowseError { get; set; }
public Dictionary<uint, AbCipUdtShape?> Shapes { get; } = [];
public Dictionary<uint, string> TemplateErrors { get; } = [];
/// <summary>Template ids that were queued but not read because <c>AB_DIFF_MAX_TEMPLATES</c> was reached.</summary>
public int TemplatesTruncated { get; set; }
public Dictionary<string, ReadObservation> Reads { get; } = new(StringComparer.Ordinal);
/// <summary>
/// libplctag only: a SECOND paced read of every selected tag, taken after the first pass and before native
/// starts (T21 press pass 1). A tag whose two reads differ is live — its value moves on its own — and is
/// compared with native structurally, not by value. Empty on the native side.
/// </summary>
public Dictionary<string, ReadObservation> ProbeReads { get; } = new(StringComparer.Ordinal);
}
/// <summary>The outcome of one whole differential run.</summary>
internal sealed class DifferentialResult
{
public required DifferentialTarget Target { get; init; }
public required TransportObservations Libplctag { get; init; }
public required TransportObservations Native { get; init; }
public required IReadOnlyList<SelectedTag> Selection { get; init; }
/// <summary>Native Forward Opens counted between the counter reset and the anchor release.</summary>
public required int NativeForwardOpens { get; init; }
/// <summary>Forward Close frames the native side put on the wire (sent) and got a reply to (acked), over the
/// same window as <see cref="NativeForwardOpens"/> — the anchor's release included (T21 review r1 M9).</summary>
public required NativeCloses NativeForwardCloses { get; init; }
/// <summary>TCP connections to the endpoint once the native session was released and drained (null when unobservable).</summary>
public required int? ConnectionsAfterNative { get; init; }
/// <summary>TCP connections to the endpoint before libplctag started, and once its handles were disposed.</summary>
public required int? ConnectionsBaseline { get; init; }
public required int? ConnectionsBeforeNative { get; init; }
public required TimeSpan LibplctagDrainWait { get; init; }
/// <summary>The endpoint's resolved addresses, and this machine's connections to it before the run (null when
/// unobservable). Reused by the latency comparison's barriers, which must not start alongside a session.</summary>
public required IReadOnlySet<IPAddress> Addresses { get; init; }
public required IReadOnlySet<IPEndPoint>? BaselineEndpoints { get; init; }
}
/// <summary>Forward Close frames the native side sent, and how many of them got a reply within their 1 s budget.</summary>
internal readonly record struct NativeCloses(int Sent, int Acked);
/// <summary>
/// The run stopped at its first connection-level failure, or when the native side opened more than once.
/// Thrown, never swallowed: every live test then fails with this message.
/// </summary>
internal sealed class DifferentialAbortedException(string message, Exception? inner = null)
: Exception(message, inner);
/// <summary>
/// Runs the whole comparison ONCE: every libplctag operation first (handles disposed), then — only
/// once libplctag's session is gone — every native operation under one anchor lease. Reads only:
/// browse listings, template reads and tag reads. Nothing on this path encodes or sends a write.
/// <para><b>Stop on first failure (review r1 C1).</b> The first connection-level failure on either side —
/// connect / RegisterSession / Forward Open refused, timeout, session lost — aborts the whole run with the
/// operation and status; nothing is retried. The native side is additionally capped at ONE open attempt
/// process-wide (<see cref="CipConnection.OpenAttemptBudget"/>), so a reconnect cannot reach the wire even
/// from inside an enumerator that swallows a per-program failure, and the run aborts the moment
/// <see cref="CipConnection.ForwardOpenCount"/> passes 1.</para>
/// </summary>
internal static class DifferentialRun
{
private const int MaxArrays = 20;
private const int MaxUdtMembers = 20;
/// <summary>Upper bound on waiting for libplctag's session to close before native work may start.</summary>
private static readonly TimeSpan LibplctagDrainBudget = TimeSpan.FromSeconds(30);
/// <summary>Settle used when the platform cannot enumerate TCP connections.</summary>
private static readonly TimeSpan FallbackSettle = TimeSpan.FromSeconds(5);
/// <summary>Bound on resolving the endpoint's host name (M5).</summary>
private static readonly TimeSpan ResolveBudget = TimeSpan.FromSeconds(5);
/// <summary>
/// Statuses that describe ONE tag and leave the session healthy. Everything else — Timeout, BadConnection,
/// Abort, BadReply, Pending, any unexpected exception — is connection-level and aborts the run.
/// </summary>
private static readonly HashSet<AbStatus> TagLevelStatuses =
[
AbStatus.NotFound, AbStatus.NotAllowed, AbStatus.OutOfBounds, AbStatus.TooLarge, AbStatus.TooSmall,
AbStatus.Unsupported, AbStatus.TypeMismatch,
];
public static async Task<DifferentialResult> ExecuteAsync(DifferentialTarget target, CancellationToken ct)
{
var addresses = await ResolveAsync(target, ct).ConfigureAwait(false);
var baseline = LocalEndpointsTo(target, addresses);
// ---- Phase 1: libplctag only. Every handle is disposed before the phase ends. ----
var (lib, selection) = await RunLibplctagPhaseAsync(target, ct).ConfigureAwait(false);
// ---- Barrier: libplctag's session must be gone before the native side connects. ----
var drain = Stopwatch.StartNew();
var beforeNative = await WaitForSessionsToCloseAsync(target, addresses, baseline, "libplctag", ct).ConfigureAwait(false);
drain.Stop();
// ---- Phase 2: native only, every operation sharing ONE connection held open by the anchor. ----
var (native, opens, closes) = await RunNativePhaseAsync(target, selection, ct).ConfigureAwait(false);
// ---- Barrier: the native session must be gone too (its Forward Close is awaited in the anchor's release). ----
var afterNative = await WaitForSessionsToCloseAsync(target, addresses, baseline, "native", ct).ConfigureAwait(false);
return new DifferentialResult
{
Target = target,
Libplctag = lib,
Native = native,
Selection = selection,
NativeForwardOpens = opens,
NativeForwardCloses = closes,
ConnectionsAfterNative = afterNative,
ConnectionsBaseline = baseline?.Count,
ConnectionsBeforeNative = beforeNative,
LibplctagDrainWait = drain.Elapsed,
Addresses = addresses,
BaselineEndpoints = baseline,
};
}
/// <summary>
/// libplctag shares one session (and one Forward Open) among the handles alive at the same time, and
/// closes it when the last one is destroyed. The per-tag create/read/dispose below would therefore open
/// a fresh session per tag; an anchor handle on the listing pseudo-tag, held for the whole phase, keeps
/// every libplctag operation on ONE session — the mirror of the native anchor lease.
/// </summary>
internal static Task<(TransportObservations Observations, IReadOnlyList<SelectedTag> Selection)> RunLibplctagPhaseAsync(
DifferentialTarget target, CancellationToken ct)
{
const string phase = "libplctag";
var lib = new TransportObservations();
var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag);
return WithLibplctagAnchorAsync(phase, target, "@tags", async () =>
{
await BrowseAsync(phase, target, factories.Enumerators, lib, afterEach: null, ct).ConfigureAwait(false);
await ReadTemplatesAsync(phase, target, factories.Templates, lib, afterEach: null, ct).ConfigureAwait(false);
var selection = Select(target, lib);
await ReadTagsAsync(phase, target, factories.Tags, selection, lib.Reads, afterEach: null, ct).ConfigureAwait(false);
// Volatility probe: read every tag again, paced, so a value that moves on its own (a live analog) is
// recognised as live rather than reported as a transport divergence when native reads it later.
await ReadTagsAsync(phase, target, factories.Tags, selection, lib.ProbeReads, afterEach: null, ct).ConfigureAwait(false);
return ((TransportObservations)lib, selection);
}, ct);
}
/// <summary>
/// Holds one libplctag handle on <paramref name="anchorTag"/> for the whole of <paramref name="body"/>, so every
/// libplctag operation in it shares ONE session and ONE Forward Open. A failure to create the anchor aborts
/// before <paramref name="body"/> runs; the anchor is disposed when <paramref name="body"/> ends, either way.
/// Shared by the differential (anchor <c>@tags</c>) and the latency comparison (anchor = its first tag).
/// </summary>
internal static async Task<T> WithLibplctagAnchorAsync<T>(
string phase, DifferentialTarget target, string anchorTag, Func<Task<T>> body, CancellationToken ct)
{
var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag);
using var anchor = factories.Tags.Create(target.Params(anchorTag));
try
{
await anchor.InitializeAsync(ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
throw Abort(phase, $"anchor session ({anchorTag})", ex);
}
return await body().ConfigureAwait(false);
}
/// <summary>The native phase alone; internal so the offline abort test can drive it against a fake endpoint.</summary>
internal static async Task<(TransportObservations Observations, int ForwardOpens, NativeCloses ForwardCloses)> RunNativePhaseAsync(
DifferentialTarget target, IReadOnlyList<SelectedTag> selection, CancellationToken ct)
{
const string phase = "native";
var native = new TransportObservations();
var factories = AbCipTransportFactories.For(AbCipTransport.Native);
var (_, forwardOpens, forwardCloses) = await WithNativeAnchorAsync(phase, target, async checkOpens =>
{
await BrowseAsync(phase, target, factories.Enumerators, native, checkOpens, ct).ConfigureAwait(false);
await ReadTemplatesAsync(phase, target, factories.Templates, native, checkOpens, ct).ConfigureAwait(false);
await ReadTagsAsync(phase, target, factories.Tags, selection, native.Reads, checkOpens, ct).ConfigureAwait(false);
return true;
}).ConfigureAwait(false);
return (native, forwardOpens, forwardCloses);
}
/// <summary>
/// Runs <paramref name="body"/> with every native operation sharing ONE connection, held open by an anchor
/// lease: the process-wide Forward Open counter is reset, the open-attempt budget is set to exactly one
/// more attempt, and <paramref name="body"/> receives a check to call after each operation that aborts the
/// moment a second Forward Open reaches the wire. The anchor is released and the budget restored in
/// <c>finally</c>. Shared by the differential and the latency comparison.
/// </summary>
/// <returns>The body's result, the Forward Opens counted between the reset and the anchor release, and the
/// Forward Closes counted between the reset and the END of the anchor release (which awaits the close).</returns>
internal static async Task<(T Result, int ForwardOpens, NativeCloses ForwardCloses)> WithNativeAnchorAsync<T>(
string phase, DifferentialTarget target, Func<Action<string>, Task<T>> body)
{
var anchorParams = target.Params("@tags");
void CheckOpens(string operation)
{
var opens = Volatile.Read(ref CipConnection.ForwardOpenCount);
if (opens > 1)
throw new DifferentialAbortedException(
$"Differential ABORTED in the {phase} phase after {operation}: {opens} Forward Open frames on the wire, " +
"expected exactly one (a reconnect, or a Large→regular fallback). Nothing was retried.");
}
Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0);
Interlocked.Exchange(ref CipConnection.ForwardClosesSent, 0);
Interlocked.Exchange(ref CipConnection.ForwardClosesAcked, 0);
Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1);
T result;
int forwardOpens;
NativeCloses forwardCloses;
try
{
// Acquire does no I/O; the key and options carry no tag-specific field, so every lease shares this entry.
var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams));
try
{
result = await body(CheckOpens).ConfigureAwait(false);
}
finally
{
forwardOpens = Volatile.Read(ref CipConnection.ForwardOpenCount);
await anchor.DisposeAsync().ConfigureAwait(false);
forwardCloses = new NativeCloses(
Volatile.Read(ref CipConnection.ForwardClosesSent), Volatile.Read(ref CipConnection.ForwardClosesAcked));
}
}
finally
{
Volatile.Write(ref CipConnection.OpenAttemptBudget, -1);
}
return (result, forwardOpens, forwardCloses);
}
private static async Task BrowseAsync(
string phase, DifferentialTarget target, IAbCipTagEnumeratorFactory factory, TransportObservations into,
Action<string>? afterEach, CancellationToken ct)
{
await PaceAsync(target, ct).ConfigureAwait(false);
using var enumerator = factory.Create();
try
{
await foreach (var tag in enumerator.EnumerateAsync(target.Params("@tags"), ct).ConfigureAwait(false))
{
into.Browse.Add(tag);
afterEach?.Invoke($"browse entry '{tag.Name}'");
}
}
catch (Exception ex) when (ex is not DifferentialAbortedException && (ex is not OperationCanceledException || !ct.IsCancellationRequested))
{
if (IsConnectionLevel(ex)) throw Abort(phase, "browse (@tags)", ex);
into.BrowseError = Describe(ex);
}
afterEach?.Invoke("browse");
}
/// <summary>Reads every template id the browse surfaced, then every nested id those shapes name.</summary>
private static async Task ReadTemplatesAsync(
string phase, DifferentialTarget target, IAbCipTemplateReaderFactory factory, TransportObservations into,
Action<string>? afterEach, CancellationToken ct)
{
var queue = new Queue<uint>(into.Browse
.Where(t => t.TemplateInstanceId is not null)
.Select(t => t.TemplateInstanceId!.Value)
.Distinct()
.Order());
var seen = new HashSet<uint>(queue);
using var reader = factory.Create();
while (queue.Count > 0)
{
if (into.Shapes.Count + into.TemplateErrors.Count >= target.MaxTemplates)
{
into.TemplatesTruncated = queue.Count;
break;
}
var id = queue.Dequeue();
await PaceAsync(target, ct).ConfigureAwait(false);
try
{
var shape = CipTemplateObjectDecoder.Decode(await reader.ReadAsync(target.Params($"@udt/{id}"), id, ct).ConfigureAwait(false));
into.Shapes[id] = shape;
foreach (var nested in shape?.Members ?? [])
if (nested.NestedTemplateId is { } n && seen.Add(n)) queue.Enqueue(n);
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
if (IsConnectionLevel(ex)) throw Abort(phase, $"template {id}", ex);
into.TemplateErrors[id] = Describe(ex);
}
afterEach?.Invoke($"template {id}");
}
}
/// <summary>
/// The tags to read, chosen from the libplctag browse (the run's first phase), so both transports
/// read exactly the same list. Without <c>AB_DIFF_TAGS</c> the three kinds are taken ROUND-ROBIN
/// (array, UDT member, scalar, ...) under their quotas — up to 20 arrays and 20 UDT members — so a
/// press with thousands of scalars still exercises arrays and UDT members at any cap (review r1 I2).
/// </summary>
internal static IReadOnlyList<SelectedTag> Select(DifferentialTarget target, TransportObservations browsed)
{
var byName = new Dictionary<string, SelectedTag>(StringComparer.Ordinal);
var scalars = new List<SelectedTag>();
var arrays = new List<SelectedTag>();
var members = new List<SelectedTag>();
foreach (var tag in browsed.Browse.Where(t => !t.IsSystemTag))
{
var name = tag.ProgramScope is null ? tag.Name : $"Program:{tag.ProgramScope}.{tag.Name}";
var type = tag.DataType;
AbCipUdtShape? shape = null;
if (type == AbCipDataType.Structure)
{
shape = tag.TemplateInstanceId is { } id ? browsed.Shapes.GetValueOrDefault(id) : null;
// A string-family struct (STRING, STR0016, ...) is read and presented as one String value.
if (shape is { IsStringFamily: true }) type = AbCipDataType.String;
}
if (type != AbCipDataType.Structure)
{
var selected = new SelectedTag(name, type, tag.IsArray ? tag.ElementCount : 1, tag.IsArray,
tag.IsArray ? "array" : "scalar");
byName.TryAdd(name, selected);
(tag.IsArray ? arrays : scalars).Add(selected);
continue;
}
if (tag.IsArray || shape is null) continue;
foreach (var member in shape.Members)
{
var memberType = member.DataType;
if (memberType == AbCipDataType.Structure)
{
var nested = member.NestedTemplateId is { } n ? browsed.Shapes.GetValueOrDefault(n) : null;
if (nested is not { IsStringFamily: true }) continue;
memberType = AbCipDataType.String;
}
// Hidden host bytes of packed BOOL members ("ZZZZZZZZZZ..." in Logix) are not addressable.
if (member.Name.StartsWith("ZZZZZZZZZZ", StringComparison.Ordinal)) continue;
var isArray = member.ArrayLength > 1;
var selected = new SelectedTag($"{name}.{member.Name}", memberType, isArray ? member.ArrayLength : 1,
isArray, "udt member");
byName.TryAdd(selected.Name, selected);
members.Add(selected);
}
}
if (target.Tags is { } explicitNames)
return explicitNames
.Select(n => byName.GetValueOrDefault(n) ?? new SelectedTag(n, null, 1, false, "explicit (type unknown)"))
.DistinctBy(t => t.Name).Take(target.MaxTags).ToList();
var chosen = new List<SelectedTag>();
var seen = new HashSet<string>(StringComparer.Ordinal);
// Reference-type enumerators created ONCE: a boxed List<T>.Enumerator struct would restart per round.
IEnumerator<SelectedTag>[] sources =
[
arrays.Take(MaxArrays).GetEnumerator(),
members.Take(MaxUdtMembers).GetEnumerator(),
scalars.AsEnumerable().Take(int.MaxValue).GetEnumerator(),
];
try
{
var live = true;
while (live && chosen.Count < target.MaxTags)
{
live = false;
foreach (var source in sources)
{
if (chosen.Count >= target.MaxTags) break;
while (source.MoveNext())
{
live = true;
if (seen.Add(source.Current.Name)) { chosen.Add(source.Current); break; }
}
}
}
}
finally
{
foreach (var source in sources) source.Dispose();
}
return chosen;
}
/// <summary>
/// Coverage problems of a default (browse-driven) selection: browse offered arrays or UDT members but
/// none were selected. Empty for an explicit <c>AB_DIFF_TAGS</c> list, which is the operator's choice.
/// </summary>
internal static List<string> SelectionCoverageProblems(
DifferentialTarget target, TransportObservations browsed, IReadOnlyList<SelectedTag> selection)
{
var problems = new List<string>();
if (target.Tags is not null) return problems;
var available = Select(target with { MaxTags = DifferentialTarget.Ceiling }, browsed);
foreach (var origin in new[] { "array", "udt member" })
if (available.Any(t => t.Origin == origin) && selection.All(t => t.Origin != origin))
problems.Add($"browse offered {origin}s but the selection holds none");
return problems;
}
private static async Task ReadTagsAsync(
string phase, DifferentialTarget target, IAbCipTagFactory factory, IReadOnlyList<SelectedTag> selection,
Dictionary<string, ReadObservation> into, Action<string>? afterEach, CancellationToken ct)
{
foreach (var tag in selection)
{
ct.ThrowIfCancellationRequested();
var runtime = factory.Create(target.Params(tag.Name, tag.ElementCount, tag.IsArray, tag.Type));
try
{
string? error = null;
try
{
await PaceAsync(target, ct).ConfigureAwait(false);
await runtime.InitializeAsync(ct).ConfigureAwait(false);
await PaceAsync(target, ct).ConfigureAwait(false);
await runtime.ReadAsync(ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
if (IsConnectionLevel(ex)) throw Abort(phase, $"tag '{tag.Name}'", ex);
error = Describe(ex);
}
var status = runtime.GetStatus();
byte[]? raw = null;
object? decoded = null;
string? decodeError = null;
if (error is null)
{
try
{
raw = RawBufferOf(runtime).ToArray();
if (tag.Type is { } type)
decoded = tag.IsArray ? runtime.DecodeArray(type, tag.ElementCount) : runtime.DecodeValue(type, null);
}
catch (Exception ex)
{
decodeError = Describe(ex); // recorded, and a NO verdict (review r1 M3)
}
}
into[tag.Name] = new ReadObservation(status, error, raw, decoded, decodeError,
(runtime as NativeAbCipTagRuntime)?.TypeCode);
}
finally
{
runtime.Dispose();
}
afterEach?.Invoke($"tag '{tag.Name}'");
}
}
private static ReadOnlyMemory<byte> RawBufferOf(IAbCipTagRuntime runtime) => runtime switch
{
LibplctagTagRuntime l => l.RawBuffer,
NativeAbCipTagRuntime n => n.RawBuffer,
_ => throw new InvalidOperationException($"Unexpected runtime {runtime.GetType().Name}."),
};
/// <summary>True when the failure is about the session, not one tag: the run must stop.</summary>
internal static bool IsConnectionLevel(Exception ex) => ex switch
{
AbTransportException ate => !TagLevelStatuses.Contains(ate.Status),
// Local, pre-I/O refusals of one tag name (malformed path, unsupported type): no session involved.
ArgumentException or FormatException or NotSupportedException => false,
_ => true,
};
internal static DifferentialAbortedException Abort(string phase, string operation, Exception ex) =>
new($"Differential ABORTED in the {phase} phase at {operation}: {Describe(ex)}. " +
"Stopped on the first connection-level failure; nothing was retried.", ex);
internal static Task PaceAsync(DifferentialTarget target, CancellationToken ct) =>
target.PaceMs > 0 ? Task.Delay(target.PaceMs, ct) : Task.CompletedTask;
/// <summary>Resolves the endpoint once, bounded (an IP literal needs no lookup).</summary>
private static async Task<IReadOnlySet<IPAddress>> ResolveAsync(DifferentialTarget target, CancellationToken ct)
{
if (IPAddress.TryParse(target.Host, out var literal)) return new HashSet<IPAddress> { literal.MapToIPv6() };
using var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct);
bounded.CancelAfter(ResolveBudget);
try
{
var resolved = await Dns.GetHostAddressesAsync(target.Host, bounded.Token).ConfigureAwait(false);
return resolved.Select(x => x.MapToIPv6()).ToHashSet();
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
throw new DifferentialAbortedException(
$"Differential ABORTED before any I/O: could not resolve '{target.Host}' within {ResolveBudget.TotalSeconds:0}s ({ex.Message}).", ex);
}
}
/// <summary>
/// A transport closes its session asynchronously (libplctag on a background thread some time after the last
/// handle is destroyed). Wait until no TCP connection to the endpoint remains that was not already there
/// before the run (compared by local endpoint, not by count — review r1 M6), so two sessions are never
/// connected to the controller at the same time.
/// </summary>
/// <param name="whose">Names the session waited for, in the abort message.</param>
/// <returns>The connection count observed when the next phase may start (null when unobservable).</returns>
internal static async Task<int?> WaitForSessionsToCloseAsync(
DifferentialTarget target, IReadOnlySet<IPAddress> addresses, IReadOnlySet<IPEndPoint>? baseline, string whose,
CancellationToken ct)
{
if (baseline is null)
{
await Task.Delay(FallbackSettle, ct).ConfigureAwait(false);
return null;
}
var deadline = Stopwatch.StartNew();
while (true)
{
var now = LocalEndpointsTo(target, addresses);
if (now is null) return null;
var added = now.Where(e => !baseline.Contains(e)).ToList();
if (added.Count == 0) return now.Count;
if (deadline.Elapsed > LibplctagDrainBudget)
throw new DifferentialAbortedException(
$"{whose}'s session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after it was " +
$"released (local {string.Join(", ", added)}); refusing to start the next phase alongside it.");
await Task.Delay(TimeSpan.FromMilliseconds(250), ct).ConfigureAwait(false);
}
}
/// <summary>Local endpoints of this machine's open TCP connections to the target, or null when the OS cannot say.</summary>
private static IReadOnlySet<IPEndPoint>? LocalEndpointsTo(DifferentialTarget target, IReadOnlySet<IPAddress> addresses)
{
try
{
return IPGlobalProperties.GetIPGlobalProperties().GetActiveTcpConnections()
.Where(c => c.RemoteEndPoint.Port == target.Port
&& addresses.Contains(c.RemoteEndPoint.Address.MapToIPv6())
&& c.State is not (TcpState.Closed or TcpState.TimeWait or TcpState.Unknown))
.Select(c => c.LocalEndPoint)
.ToHashSet();
}
catch (Exception ex) when (ex is PlatformNotSupportedException or NetworkInformationException or NotImplementedException)
{
return null;
}
}
/// <summary>Order-insensitive multiset difference: entries of <paramref name="a"/> missing from <paramref name="b"/>.</summary>
internal static List<T> Missing<T>(IEnumerable<T> a, IEnumerable<T> b) where T : notnull
{
var counts = b.GroupBy(x => x).ToDictionary(g => g.Key, g => g.Count());
var missing = new List<T>();
foreach (var x in a)
{
if (counts.TryGetValue(x, out var c) && c > 0) counts[x] = c - 1;
else missing.Add(x);
}
return missing;
}
/// <summary>Structural equality for decoded values: arrays element-wise, NaN equal to NaN.</summary>
internal static bool ValuesEqual(object? a, object? b) =>
a is Array || b is Array
? StructuralComparisons.StructuralEqualityComparer.Equals(a, b)
: Equals(a, b);
/// <summary>
/// True when the libplctag volatility probe saw the value move by itself: both libplctag reads succeeded and
/// their bytes differ. A missing or failed probe is never live, so such a tag keeps the full value comparison.
/// </summary>
internal static bool IsLive(ReadObservation lib, ReadObservation? probe) =>
probe is not null && lib.Succeeded && probe.Succeeded && lib.Raw is not null && probe.Raw is not null
&& !lib.Raw.AsSpan().SequenceEqual(probe.Raw);
/// <summary>
/// The value check that survives for a LIVE tag (T21 review r1 I1). A scalar is one element, an array element-wise
/// over the decoded values:
/// <list type="bullet">
/// <item>an element EQUAL in both libplctag reads did not move, so native must read it equal too —
/// otherwise <c>**NO: stable element i**</c> (a stride, offset or byte-order defect lands here);</item>
/// <item>an element that moved is excused its value, but native must not read a NaN/∞ that neither
/// libplctag read saw — <c>**NO: non-finite element i**</c>.</item>
/// </list>
/// Null means the check passed. Shapes must already agree between libplctag and native.
/// </summary>
internal static string? LiveElementMismatch(object? lib, object? probe, object? nat)
{
var l = Elements(lib);
var p = Elements(probe);
var n = Elements(nat);
if (l.Length != p.Length) return "**NO: probe shape**"; // libplctag disagreed with itself: never excuse it
if (l.Length != n.Length) return "**NO: shape**";
for (var i = 0; i < l.Length; i++)
{
if (Equals(l[i], p[i]))
{
if (!Equals(l[i], n[i])) return $"**NO: stable element {i}**";
}
else if (!IsFinite(n[i]) && !Equals(n[i], l[i]) && !Equals(n[i], p[i]))
{
return $"**NO: non-finite element {i}**";
}
}
return null;
}
private static object?[] Elements(object? v) => v is Array a ? a.Cast<object?>().ToArray() : [v];
private static bool IsFinite(object? v) => v switch
{
float f => float.IsFinite(f),
double d => double.IsFinite(d),
_ => true,
};
/// <summary>The same decoded shape: both scalars of one CLR type, or arrays of one element type and length.</summary>
internal static bool SameShape(object? a, object? b) => (a, b) switch
{
(null, null) => true,
(Array x, Array y) => x.GetType() == y.GetType() && x.Length == y.Length,
(Array, _) or (_, Array) => false,
(not null, not null) => a.GetType() == b.GetType(),
_ => false,
};
/// <summary>Equal only when BOTH decoded: an undecodable shape is never parity (review r1 I3).</summary>
internal static bool ShapesEqual(AbCipUdtShape? a, AbCipUdtShape? b) =>
a is not null && b is not null
&& a.TypeName == b.TypeName && a.TotalSize == b.TotalSize && a.Members.SequenceEqual(b.Members);
/// <summary>
/// Template-parity problems; empty means parity. Vacuity is a problem: browse found structured tags
/// but no template id was compared (review r1 I3).
/// </summary>
internal static List<string> TemplateProblems(TransportObservations lib, TransportObservations native)
{
var problems = new List<string>();
var ids = lib.Shapes.Keys.Concat(lib.TemplateErrors.Keys)
.Concat(native.Shapes.Keys).Concat(native.TemplateErrors.Keys)
.Distinct().Order().ToList();
var structured = lib.Browse.Concat(native.Browse).Count(t => t.DataType == AbCipDataType.Structure);
if (structured > 0 && ids.Count == 0)
problems.Add($"vacuous: browse found {structured} structured tag(s) but no template id was compared");
foreach (var id in ids)
{
var libShape = lib.Shapes.GetValueOrDefault(id);
var nativeShape = native.Shapes.GetValueOrDefault(id);
if (!ShapesEqual(libShape, nativeShape))
problems.Add($"template {id}: " + (libShape is null || nativeShape is null ? "undecodable or unread on a side" : "shapes differ"));
}
return problems;
}
internal static string Describe(Exception ex) => ex is AbTransportException ate
? $"{ate.Status}: {ate.Message}"
: $"{ex.GetType().Name}: {ex.Message}";
}