fix(opcuaclient): #676 live gate A1 — a sustained closed connection is a definitive loss
Live gate A1 (docker-dev, opc-plc restart / SIGINT / kill): the fast path never fired — a real process stop delivers no session status; the keep-alives report BadSecureChannelClosed (0x80860000) once, then BadConnectionClosed (0x80AE0000) interleaved with BadNoCommunication for the whole outage. Reproduced in-process with TcpStallProxy.HardClose (reset every connection, refuse new ones; server and session alive behind it): the identical sequence, and the SDK 1.5.378 session never opens a new connection on its own (48 s with the proxy accepting again, zero attempts) — a closed channel cannot recover in place. The driver now treats BadSecureChannelClosed/BadConnectionClosed as definitive on the SECOND closed-channel keep-alive failure in one outage (a good keep-alive resets the count), raising ConnectionLost ~1 keep-alive after the drop. BadNoCommunication / BadRequestTimeout / BadNotConnected (stall/pause) stay with the watchdog; the stall test still sees no ConnectionLost. New wire test (hard close + reopen ⇒ raised once, SDK never reconnects, reinit brings the identical value back Good) and unit tests (sustained raises once; a single blip that recovers does not) — all red on the previous code. Docs updated.
This commit is contained in:
Joseph Doherty
+48
-2
@@ -105,11 +105,9 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
public static TheoryData<uint> AmbiguousStatuses =>
|
||||
[
|
||||
StatusCodes.BadNoCommunication,
|
||||
StatusCodes.BadConnectionClosed,
|
||||
StatusCodes.BadNotConnected,
|
||||
StatusCodes.BadRequestTimeout,
|
||||
StatusCodes.BadTimeout,
|
||||
StatusCodes.BadSecureChannelClosed,
|
||||
];
|
||||
|
||||
[Theory]
|
||||
@@ -153,6 +151,54 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
raised.ShouldBe(0, "an ambiguous loss may yet resume in place — it belongs to the #527 watchdog");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #676 live gate A1: a real upstream process stop (docker restart / SIGINT / docker kill) never produced a
|
||||
/// session status — the keep-alives carried <c>BadSecureChannelClosed</c> once, then <c>BadConnectionClosed</c>
|
||||
/// (interleaved with <c>BadNoCommunication</c>) for the whole outage. Measured in-process (hard-closing proxy, the
|
||||
/// server and its session still alive): the SDK 1.5.378 session NEVER opens a new connection on its own, so a
|
||||
/// closed channel is dead for good. Sustained — a second closed-channel failure in the same outage — it raises.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_sustained_closed_channel_raises_ConnectionLost_once()
|
||||
{
|
||||
var (drv, _, _, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
var raised = new List<string>();
|
||||
drv.ConnectionLost += (_, e) => raised.Add(e.Reason);
|
||||
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadSecureChannelClosed); // the live sequence …
|
||||
raised.ShouldBeEmpty("one closed-channel failure may be a transient");
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadConnectionClosed); // … sustained
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadConnectionClosed);
|
||||
|
||||
raised.ShouldHaveSingleItem().ShouldContain("BadConnectionClosed");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0x80860000u)] // BadSecureChannelClosed
|
||||
[InlineData(0x80AE0000u)] // BadConnectionClosed
|
||||
public void A_single_closed_channel_blip_that_recovers_does_not_raise(uint closed)
|
||||
{
|
||||
var (drv, _, _, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
var raised = 0;
|
||||
drv.ConnectionLost += (_, _) => raised++;
|
||||
|
||||
drv.RaiseKeepAliveForTest(closed);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good); // recovered — the count starts again
|
||||
drv.RaiseKeepAliveForTest(closed);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
|
||||
raised.ShouldBe(0);
|
||||
// Positive control: the same driver DOES raise once a closed channel is sustained.
|
||||
drv.RaiseKeepAliveForTest(closed);
|
||||
drv.RaiseKeepAliveForTest(closed);
|
||||
raised.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_ambiguous_outage_that_turns_definitive_raises_once()
|
||||
{
|
||||
|
||||
+72
@@ -156,4 +156,76 @@ public sealed class OpcUaClientDriverStallWireTests
|
||||
Volatile.Read(ref connectionLost).ShouldBe(0,
|
||||
"a stall is AMBIGUOUS (the session survived it) — it must go through the #527 watchdog, never the fast path");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #676 live gate A1 — a real upstream process stop is a CLOSED connection, not a stall: the keep-alives report
|
||||
/// <c>BadSecureChannelClosed</c>, then <c>BadConnectionClosed</c> (with <c>BadNoCommunication</c>) for the rest of
|
||||
/// the outage, and the SDK never opens a new connection on its own — not even when the server (here, with the
|
||||
/// session still alive behind the proxy) is reachable again. The driver must therefore raise
|
||||
/// <c>ConnectionLost</c> once the closed channel is sustained, so the host re-initialises in seconds instead of
|
||||
/// waiting ~90 s for the watchdog; the re-initialise brings the identical value back Good.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_hard_closed_connection_raises_ConnectionLost_once_and_the_reinit_recovers_in_seconds()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-hardclose", ct);
|
||||
await using var proxy = TcpStallProxy.Start(new Uri(rig.EndpointUrl).Port);
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
|
||||
|
||||
await using var drv = OpcUaClientDriverWireTests.NewDriver(
|
||||
rig, "wire-hardclose", endpointUrl: proxy.Through(rig.EndpointUrl), sessionTimeout: TimeSpan.FromSeconds(120));
|
||||
var lost = new List<string>();
|
||||
drv.ConnectionLost += (_, e) => { lock (lost) lost.Add(e.Reason); };
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
var seen = new List<Seen>();
|
||||
drv.OnDataChange += (_, e) =>
|
||||
{
|
||||
if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef)
|
||||
lock (seen) seen.Add(new Seen(e.Snapshot.StatusCode, e.Snapshot.Value, e.Snapshot.SourceTimestampUtc));
|
||||
};
|
||||
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
|
||||
await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 10.0f }); }, "initial value");
|
||||
var session = drv.Session.ShouldNotBeNull();
|
||||
var keepAlives = new List<uint>();
|
||||
session.KeepAlive += (_, e) => { lock (keepAlives) keepAlives.Add(e.Status?.StatusCode.Code ?? 0); };
|
||||
|
||||
// ── The upstream process goes away: every connection reset, new ones refused ──────────────────────────────
|
||||
var killedAt = DateTime.UtcNow;
|
||||
proxy.HardClose();
|
||||
await WaitAsync(() => { lock (lost) return lost.Count > 0; },
|
||||
"a sustained closed channel must raise ConnectionLost", TimeSpan.FromSeconds(15));
|
||||
(DateTime.UtcNow - killedAt).ShouldBeLessThan(TimeSpan.FromSeconds(15), "seconds, not the watchdog's ~90 s");
|
||||
lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "the outage was published Bad first");
|
||||
lock (keepAlives)
|
||||
{
|
||||
keepAlives.ShouldContain(0x80860000u, "BadSecureChannelClosed — the live-gate sequence");
|
||||
keepAlives.ShouldContain(0x80AE0000u, "BadConnectionClosed — the live-gate sequence");
|
||||
}
|
||||
|
||||
// The server is back — the session still exists on it — yet the SDK never reconnects by itself: no new
|
||||
// connection reaches the proxy while further keep-alives fail. That is what makes the closed channel definitive.
|
||||
proxy.Reopen();
|
||||
var accepted = proxy.AcceptedConnections;
|
||||
int kaAtReopen;
|
||||
lock (keepAlives) kaAtReopen = keepAlives.Count;
|
||||
await WaitAsync(() => { lock (keepAlives) return keepAlives.Count >= kaAtReopen + 4; }, "keep-alives keep failing");
|
||||
proxy.AcceptedConnections.ShouldBe(accepted, "the SDK session never reopens a closed connection on its own");
|
||||
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy);
|
||||
lock (lost) lost.Count.ShouldBe(1, "once per session, however many closed-channel keep-alives follow");
|
||||
|
||||
// ── The host's response (DriverInstanceActor: Reconnecting → ShutdownAsync + InitializeAsync → re-subscribe) ──
|
||||
await drv.ShutdownAsync(ct);
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
int before;
|
||||
lock (seen) before = seen.Count;
|
||||
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
|
||||
await WaitAsync(() => { lock (seen) return seen.Skip(before).Any(s => s is { Status: 0u, Value: 10.0f }); },
|
||||
"the identical value must come back Good after the re-initialise");
|
||||
(DateTime.UtcNow - killedAt).ShouldBeLessThan(TimeSpan.FromSeconds(60), "well under the watchdog's 90 s");
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 77.0f);
|
||||
await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 77.0f }); }, "a later change arrives");
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
|
||||
lock (lost) lost.Count.ShouldBe(1, "the healthy re-initialised session reports no loss");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ internal sealed class TcpStallProxy : IAsyncDisposable
|
||||
private readonly List<TcpClient> _sockets = [];
|
||||
private readonly Lock _lock = new();
|
||||
private TaskCompletionSource _gate = Open();
|
||||
private volatile bool _refusing;
|
||||
private Task _acceptLoop = Task.CompletedTask;
|
||||
|
||||
private TcpStallProxy(int targetPort) => _targetPort = targetPort;
|
||||
@@ -48,6 +49,34 @@ internal sealed class TcpStallProxy : IAsyncDisposable
|
||||
if (_gate.Task.IsCompleted) _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The shape of a killed / stopped server process (#676 live gate): every open connection is RESET (linger 0 ⇒
|
||||
/// RST, both sides) and new connections are refused (accepted and reset at once) until <see cref="Reopen"/>.
|
||||
/// </summary>
|
||||
public void HardClose()
|
||||
{
|
||||
List<TcpClient> victims;
|
||||
lock (_lock)
|
||||
{
|
||||
_refusing = true;
|
||||
victims = [.. _sockets];
|
||||
_sockets.Clear();
|
||||
}
|
||||
foreach (var s in victims) Reset(s);
|
||||
}
|
||||
|
||||
/// <summary>Accept and forward new connections again (the server is back); the reset ones stay dead.</summary>
|
||||
public void Reopen() => _refusing = false;
|
||||
|
||||
/// <summary>Connections refused while <see cref="HardClose"/> was in force — a reconnect ATTEMPT by the client.</summary>
|
||||
public int RefusedConnections { get; private set; }
|
||||
|
||||
private static void Reset(TcpClient s)
|
||||
{
|
||||
try { s.Client.LingerState = new LingerOption(true, 0); } catch { }
|
||||
try { s.Dispose(); } catch { }
|
||||
}
|
||||
|
||||
/// <summary>Releases everything held since <see cref="Pause"/>, in order.</summary>
|
||||
public void Resume()
|
||||
{
|
||||
@@ -73,6 +102,12 @@ internal sealed class TcpStallProxy : IAsyncDisposable
|
||||
TcpClient client;
|
||||
try { client = await _listener.AcceptTcpClientAsync(_cts.Token); }
|
||||
catch { return; }
|
||||
if (_refusing)
|
||||
{
|
||||
lock (_lock) RefusedConnections++;
|
||||
Reset(client);
|
||||
continue;
|
||||
}
|
||||
var upstream = new TcpClient();
|
||||
try { await upstream.ConnectAsync(IPAddress.Loopback, _targetPort, _cts.Token); }
|
||||
catch
|
||||
|
||||
Reference in New Issue
Block a user