Joseph Doherty
f0992394b6
RegisterNotificationAsync published a refused registration into
NativeRegistrations under ConnectGate with NextRetryUtc = MinValue; the
SubscribeAsync caller booked the refusal backoff (NoteDeadFailure refused:
true) only after the gate was released and after its WARN. A probe tick in
that window took the gate, saw the registration as due, and retried it at
once - the CI flake in A_rearm_during_an_in_flight_retry_is_not_lost
("AttemptsFor(\"MAIN.T\") should be 1 but was 2").
RegisterNotificationAsync now books the failure (refusal backoff, or the
comms-class every-tick retry) before it publishes the dead registration, still
under the gate. The caller no longer books the refusal a second time (that
would double the backoff); its comms-class booking is idempotent and stays.
Tests: TwinCATCapturingLogger gains an OnWarning hook. A logger that holds
the refusal WARN for ten probe intervals forces the window open, so
A_rearm_during_an_in_flight_retry_is_not_lost exercises it on every run, and
the new A_refused_registration_is_published_with_its_backoff_already_set
pins the backoff (booked once, not doubled).
Pass rate of A_rearm_during_an_in_flight_retry_is_not_lost:
- before, unmodified test: 40/40 runs, plus 200/200 in-process repeats under
CPU load (the natural window is a few instructions wide; the flake is rare);
- before, window held open: 0/32 (deterministic failure, "should be 1 but
was 2");
- after, window held open: 40/40.
Full TwinCAT suite 663/663 at AKKA_TEST_TIMEFACTOR 1 and 3.
Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
308 lines
17 KiB
C#
308 lines
17 KiB
C#
using System.Collections.Concurrent;
|
|
using Shouldly;
|
|
using TwinCAT.TypeSystem;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
|
|
|
|
/// <summary>
|
|
/// #739 review fix round 2: a cancelled shutdown's background teardown is confined to its own generation
|
|
/// (I1); write outcomes feed per-device health (M1); a symbol-version re-arm survives an in-flight retry
|
|
/// and restarts the backoff (M2/M3); WSTRING / multi-dimensional encode plans (M5); infinity strings (M6).
|
|
/// </summary>
|
|
[Trait("Category", "Unit")]
|
|
public sealed class TwinCATReviewFixRound2Tests
|
|
{
|
|
private const string Host = "ads://5.23.91.23.1.1:851";
|
|
|
|
private static TwinCATProbeOptions FastProbe => new()
|
|
{
|
|
Enabled = true,
|
|
Interval = TimeSpan.FromMilliseconds(30),
|
|
Timeout = TimeSpan.FromMilliseconds(100),
|
|
};
|
|
|
|
private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, TwinCATProbeOptions probe, bool native,
|
|
params TwinCATTagDefinition[] tags) =>
|
|
NewDriver(factory, probe, native, logger: null, tags);
|
|
|
|
private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, TwinCATProbeOptions probe, bool native,
|
|
TwinCATCapturingLogger? logger, params TwinCATTagDefinition[] tags) =>
|
|
new(new TwinCATDriverOptions
|
|
{
|
|
Devices = [new TwinCATDeviceOptions(Host)],
|
|
RawTags = TwinCATRawTags.Entries(tags),
|
|
Probe = probe,
|
|
UseNativeNotifications = native,
|
|
}, "drv-1", factory, logger);
|
|
|
|
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
|
|
{
|
|
var deadline = DateTime.UtcNow + timeout;
|
|
while (!condition() && DateTime.UtcNow < deadline)
|
|
await Task.Delay(10);
|
|
}
|
|
|
|
// ---- I1 ----
|
|
|
|
/// <summary>
|
|
/// The exact reviewed sequence: a poll read is wedged; ShutdownAsync is cancelled (its teardown
|
|
/// continues in the background); the same instance is re-initialised and subscribed; the wedge is
|
|
/// released and the OLD teardown completes. The NEW subscription keeps polling and delivering (positive
|
|
/// control), UnsubscribeAsync finds and stops it, and a final shutdown leaves nothing polling.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_cancelled_shutdowns_background_teardown_never_touches_the_next_generations_subscription()
|
|
{
|
|
var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var counter = 0;
|
|
var clients = new ConcurrentQueue<FakeTwinCATClient>();
|
|
var factory = new FakeTwinCATClientFactory
|
|
{
|
|
Customise = () =>
|
|
{
|
|
// Client 1 (generation 1) wedges its first read; later clients answer an ever-changing value
|
|
// so every poll publishes.
|
|
var c = new ChangingValueFake(() => Interlocked.Increment(ref counter))
|
|
{
|
|
BlockRead = clients.IsEmpty ? wedge : null,
|
|
};
|
|
clients.Enqueue(c);
|
|
return c;
|
|
},
|
|
};
|
|
var tag = new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt);
|
|
var drv = NewDriver(factory, new TwinCATProbeOptions { Enabled = false }, native: false, tag);
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
var events = new ConcurrentQueue<DataChangeEventArgs>();
|
|
drv.OnDataChange += (_, e) => events.Enqueue(e);
|
|
|
|
_ = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken);
|
|
await WaitForAsync(() => !clients.IsEmpty, TimeSpan.FromSeconds(2)); // gen-1 poll read is wedged
|
|
|
|
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)))
|
|
await Should.ThrowAsync<OperationCanceledException>(() => drv.ShutdownAsync(cts.Token));
|
|
var oldTeardown = drv.LastTeardownForTest.ShouldNotBeNull();
|
|
oldTeardown.IsCompleted.ShouldBeFalse("the old teardown must still be running in the background");
|
|
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
var handle = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken);
|
|
|
|
wedge.SetResult();
|
|
await oldTeardown.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
|
|
|
// Positive control: the new subscription delivers AFTER the old teardown finished.
|
|
var afterTeardown = events.Count(e => e.SubscriptionHandle.Equals(handle));
|
|
await WaitForAsync(() => events.Count(e => e.SubscriptionHandle.Equals(handle)) >= afterTeardown + 3, TimeSpan.FromSeconds(3));
|
|
events.Count(e => e.SubscriptionHandle.Equals(handle)).ShouldBeGreaterThanOrEqualTo(afterTeardown + 3);
|
|
|
|
// UnsubscribeAsync can still find and stop it.
|
|
await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
|
|
var stoppedAt = events.Count(e => e.SubscriptionHandle.Equals(handle));
|
|
// Measured window: a sibling driver's poll publishes 5 times; the stopped loop must publish nothing.
|
|
var witnessFactory = new FakeTwinCATClientFactory { Customise = () => new ChangingValueFake(() => Environment.TickCount) };
|
|
var witnessDrv = NewDriver(witnessFactory, new TwinCATProbeOptions { Enabled = false }, native: false, tag);
|
|
await witnessDrv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
var witnessEvents = 0;
|
|
witnessDrv.OnDataChange += (_, _) => Interlocked.Increment(ref witnessEvents);
|
|
_ = await witnessDrv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken);
|
|
await WaitForAsync(() => Volatile.Read(ref witnessEvents) >= 5, TimeSpan.FromSeconds(3));
|
|
Volatile.Read(ref witnessEvents).ShouldBeGreaterThanOrEqualTo(5);
|
|
await witnessDrv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
|
|
events.Count(e => e.SubscriptionHandle.Equals(handle)).ShouldBe(stoppedAt, "the unsubscribed loop must have stopped");
|
|
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
/// <summary>Answers each read with a new value so every poll cycle publishes.</summary>
|
|
private sealed class ChangingValueFake(Func<int> next) : FakeTwinCATClient
|
|
{
|
|
public override Task<(object? value, uint status)> ReadValueAsync(
|
|
string symbolPath, TwinCATDataType type, int? bitIndex, int? arrayCount, CancellationToken ct)
|
|
{
|
|
if (BlockRead is not null) return base.ReadValueAsync(symbolPath, type, bitIndex, arrayCount, ct);
|
|
Values[symbolPath] = next();
|
|
return base.ReadValueAsync(symbolPath, type, bitIndex, arrayCount, ct);
|
|
}
|
|
}
|
|
|
|
// ---- M1 ----
|
|
|
|
/// <summary>
|
|
/// A write-only device with the probe off: a wire-failed write degrades; a later Good write is the device
|
|
/// answering — Healthy again with the stamp advanced. A refused write (BadNotWritable) changes nothing.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_good_write_clears_that_devices_comms_fault()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake },
|
|
new TwinCATProbeOptions { Enabled = false }, native: false,
|
|
new TwinCATTagDefinition("W", Host, "MAIN.W", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
var initStamp = drv.GetHealth().LastSuccessfulRead!.Value;
|
|
|
|
fake.ThrowOnWrite = true;
|
|
_ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken);
|
|
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
|
|
|
fake.ThrowOnWrite = false;
|
|
fake.WriteStatuses["MAIN.W"] = TwinCATStatusMapper.BadNotWritable;
|
|
_ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken);
|
|
drv.GetHealth().State.ShouldBe(DriverState.Degraded, "a refusal says nothing about the comms fault");
|
|
|
|
await WaitForAsync(() => DateTime.UtcNow > initStamp, TimeSpan.FromSeconds(1));
|
|
fake.WriteStatuses.Remove("MAIN.W");
|
|
_ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken);
|
|
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
|
|
drv.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(initStamp);
|
|
}
|
|
|
|
// ---- M2 + M3 ----
|
|
|
|
/// <summary>
|
|
/// A symbol-version re-arm that lands while a retry of the refused registration is in flight is not
|
|
/// overwritten by that retry's failure: the next tick re-attempts at once. (Without the epoch the
|
|
/// failure re-imposed a 30 s+ backoff.) The initial refusal's WARN is held for ten probe intervals so the
|
|
/// window that made this test flake (lmxopcua#773 — a tick retrying before the backoff was booked) is
|
|
/// always exercised rather than hit by chance.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_rearm_during_an_in_flight_retry_is_not_lost()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
fake.AddNotificationErrors["MAIN.T"] = new TwinCATTypeMismatchException("MAIN.T", "PLC STRING");
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, FastProbe, native: true,
|
|
HoldingRefusalWarning(),
|
|
new TwinCATTagDefinition("T", Host, "MAIN.T", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
_ = await drv.SubscribeAsync(["T"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
|
|
fake.AttemptsFor("MAIN.T").ShouldBe(1);
|
|
|
|
var inFlight = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
fake.PreAttemptGate = inFlight;
|
|
fake.FireSymbolVersionChanged(); // re-arm: the next tick starts a retry, which wedges on the gate
|
|
await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 2, TimeSpan.FromSeconds(3));
|
|
fake.FireSymbolVersionChanged(); // a second program change while that retry is in flight
|
|
fake.PreAttemptGate = null;
|
|
inFlight.SetResult(); // the in-flight retry now fails (still refused)
|
|
|
|
await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 3, TimeSpan.FromSeconds(3));
|
|
fake.AttemptsFor("MAIN.T").ShouldBeGreaterThanOrEqualTo(3);
|
|
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
/// <summary>A logger that blocks the driver's refusal WARN for ten probe intervals (see lmxopcua#773).</summary>
|
|
private static TwinCATCapturingLogger HoldingRefusalWarning()
|
|
{
|
|
var hold = TimeSpan.FromMilliseconds(FastProbe.Interval.TotalMilliseconds * 10);
|
|
return new TwinCATCapturingLogger
|
|
{
|
|
OnWarning = m => { if (m.Contains("refused native notification")) Thread.Sleep(hold); },
|
|
};
|
|
}
|
|
|
|
/// <summary>
|
|
/// lmxopcua#773: a refused registration is published to the device's registry WITH its refusal backoff
|
|
/// already set, so a probe tick can never see it with <c>NextRetryUtc = MinValue</c> and retry at once.
|
|
/// The refusal WARN is held for ten probe intervals to open the window between the subscribe path
|
|
/// releasing <c>ConnectGate</c> and its caller finishing; before the fix the backoff was booked only after
|
|
/// that warning, so a tick in the window re-attempted the tag (the CI flake in the test above:
|
|
/// <c>AttemptsFor("MAIN.T") should be 1 but was 2</c>). Deterministic: it failed every run before the fix.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_refused_registration_is_published_with_its_backoff_already_set()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
fake.AddNotificationErrors["MAIN.T"] = new TwinCATTypeMismatchException("MAIN.T", "PLC STRING");
|
|
var logger = HoldingRefusalWarning();
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, FastProbe, native: true, logger,
|
|
new TwinCATTagDefinition("T", Host, "MAIN.T", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
_ = await drv.SubscribeAsync(["T"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
|
|
|
|
logger.Warnings.ShouldContain(w => w.Contains("refused native notification")); // the window was held open
|
|
fake.AttemptsFor("MAIN.T").ShouldBe(1, "a probe tick retried the refusal before its backoff was booked");
|
|
var reg = drv.GetDeviceState(Host)!.NativeRegistrations.Values.Single();
|
|
reg.RefusalBackoff.ShouldBe(TwinCATDriver.RefusedRetryInitial, "booked once, not doubled by the caller");
|
|
reg.NextRetryUtc.ShouldBeGreaterThan(DateTime.UtcNow);
|
|
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
/// <summary>A re-arm restarts the refusal backoff from the initial delay instead of doubling the old one.</summary>
|
|
[Fact]
|
|
public async Task A_rearm_restarts_the_refusal_backoff()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
fake.AddNotificationErrors["MAIN.T"] = new TwinCATTypeMismatchException("MAIN.T", "PLC STRING");
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, FastProbe, native: true,
|
|
new TwinCATTagDefinition("T", Host, "MAIN.T", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
_ = await drv.SubscribeAsync(["T"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
|
|
var reg = drv.GetDeviceState(Host)!.NativeRegistrations.Values.Single();
|
|
reg.RefusalBackoff.ShouldBe(TwinCATDriver.RefusedRetryInitial);
|
|
|
|
fake.FireSymbolVersionChanged();
|
|
await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 2, TimeSpan.FromSeconds(3));
|
|
await WaitForAsync(() => reg.RefusalBackoff != TimeSpan.Zero, TimeSpan.FromSeconds(1));
|
|
reg.RefusalBackoff.ShouldBe(TwinCATDriver.RefusedRetryInitial);
|
|
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
// ---- M5 ----
|
|
|
|
/// <summary>A WSTRING tag over a WSTRING symbol writes UTF-16LE with an aligned 2-byte terminator, zero-padded.</summary>
|
|
[Fact]
|
|
public async Task A_WSTRING_write_goes_through_the_client_encode_plan()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
fake.SymbolShapes["MAIN.W"] = TwinCATPlcSymbolShape.Classify("WSTRING(3)", DataTypeCategory.String, 8);
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake },
|
|
new TwinCATProbeOptions { Enabled = false }, native: false,
|
|
new TwinCATTagDefinition("W", Host, "MAIN.W", TwinCATDataType.WString));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
var results = await drv.WriteAsync(
|
|
[new WriteRequest("W", "中x"), new WriteRequest("W", "abcd")], TestContext.Current.CancellationToken);
|
|
|
|
results[0].StatusCode.ShouldBe(TwinCATStatusMapper.Good);
|
|
results[1].StatusCode.ShouldBe(TwinCATStatusMapper.BadOutOfRange); // 4 chars do not fit WSTRING(3)
|
|
fake.RawWrites["MAIN.W"].ShouldBe(new byte[] { 0x2D, 0x4E, (byte)'x', 0, 0, 0, 0, 0 });
|
|
}
|
|
|
|
/// <summary>A multi-dimensional array is written flattened in PLC memory order (last index fastest).</summary>
|
|
[Fact]
|
|
public async Task A_multi_dimensional_array_write_is_flattened_row_major()
|
|
{
|
|
var fake = new FakeTwinCATClient();
|
|
var element = TwinCATPlcSymbolShape.Classify("INT", DataTypeCategory.Primitive, 2);
|
|
fake.SymbolShapes["MAIN.M"] = TwinCATPlcSymbolShape.Classify("ARRAY [0..1,0..2] OF INT", DataTypeCategory.Array, 12);
|
|
fake.ArrayShapes["MAIN.M"] = new TwinCATPlcArrayShape([2, 3], 12, element);
|
|
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake },
|
|
new TwinCATProbeOptions { Enabled = false }, native: false,
|
|
new TwinCATTagDefinition("M", Host, "MAIN.M", TwinCATDataType.Int, ArrayLength: 6));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
var result = await drv.WriteAsync([new WriteRequest("M", new[] { 1, 2, 3, 4, 5, 6 })], TestContext.Current.CancellationToken);
|
|
|
|
result.Single().StatusCode.ShouldBe(TwinCATStatusMapper.Good);
|
|
fake.RawWrites["MAIN.M"].ShouldBe(new byte[] { 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0 });
|
|
TwinCATArrayCodec.Decode(TwinCATDataType.Int, fake.RawWrites["MAIN.M"], new TwinCATArrayLayout(6, 2), false)
|
|
.ShouldBe(new short[] { 1, 2, 3, 4, 5, 6 });
|
|
}
|
|
|
|
// ---- M6 ----
|
|
|
|
/// <summary>An infinity spelled as a string passes like a float/double infinity; a finite overflow does not.</summary>
|
|
[Fact]
|
|
public void Infinity_strings_are_explicit_infinities()
|
|
{
|
|
AdsTwinCATClient.ConvertForWrite(TwinCATDataType.Real, "Infinity").ShouldBe(float.PositiveInfinity);
|
|
AdsTwinCATClient.ConvertForWrite(TwinCATDataType.LReal, "-Infinity").ShouldBe(double.NegativeInfinity);
|
|
Should.Throw<OverflowException>(() => AdsTwinCATClient.ConvertForWrite(TwinCATDataType.Real, "1e300"));
|
|
}
|
|
}
|