feat(opcuaclient): #676 follow-ups — endpoint rewrite, stall wire test, fast definitive-loss recovery
- Endpoint rewrite: the driver now connects to the configured host AND port (scheme match, advertised path kept) instead of the server-advertised URL — fixes NAT/port-map upstreams and lets an in-test proxy carry the session (SDK SelectEndpoint rule). - TcpStallProxy + OpcUaClientDriverStallWireTests: the in-place blip (#670) is now wire-tested on loopback — one Bad, frozen read stamp, same session, identical value republished Good, a post-recovery change arrives, no source-time regression after the SDK's late backlog flush. - Defect found by that test, fixed: a change published right after a stall rode a Publish response the SDK timed out and never republished, leaving the node Good on the pre-change value indefinitely. New trailing reconcile read (Timeout + KeepAliveInterval after recovery) publishes any Good value newer than the last accepted one, through the replay guard's mark. - IDriverConnectionLossSource (Core.Abstractions): opt-in definitive-loss signal. DriverInstanceActor attaches in PreStart / detaches in PostStop and turns a raise into a disconnect only while Connected. OpcUaClient raises once per session on keep-alive BadSessionIdInvalid/BadSessionClosed/BadSessionNotActivated/ BadServerHalted (measured: the SDK never re-opens a halted channel) or a dropped subscription; ambiguous codes stay with the #527 watchdog. - Replay guard: one Information summary per recovery when the window closes ("dropped N late notifications across M refs"), with a timer so quiet subscriptions log it too. - ReconnectPeriod deleted (options, AdminUI form, docs, tests); stored configs that carry reconnectPeriod still parse in the factory, probe and browser. - Docs: OpcUaClient.md, DriverLifecycle.md, driver-specs.md, CLAUDE.md.
This commit is contained in:
Joseph Doherty
-4
@@ -34,7 +34,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests
|
||||
Timeout = TimeSpan.FromSeconds(20),
|
||||
SessionTimeout = TimeSpan.FromSeconds(180),
|
||||
KeepAliveInterval = TimeSpan.FromSeconds(10),
|
||||
ReconnectPeriod = TimeSpan.FromSeconds(15),
|
||||
AutoAcceptCertificates = true,
|
||||
BrowseRoot = "i=85",
|
||||
MaxDiscoveredNodes = 5000,
|
||||
@@ -58,7 +57,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests
|
||||
back.Timeout.ShouldBe(TimeSpan.FromSeconds(20));
|
||||
back.SessionTimeout.ShouldBe(TimeSpan.FromSeconds(180));
|
||||
back.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(10));
|
||||
back.ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(15));
|
||||
back.AutoAcceptCertificates.ShouldBeTrue();
|
||||
back.BrowseRoot.ShouldBe("i=85");
|
||||
back.MaxDiscoveredNodes.ShouldBe(5000);
|
||||
@@ -105,7 +103,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests
|
||||
Timeout = TimeSpan.FromSeconds(12),
|
||||
SessionTimeout = TimeSpan.FromSeconds(150),
|
||||
KeepAliveInterval = TimeSpan.FromSeconds(7),
|
||||
ReconnectPeriod = TimeSpan.FromSeconds(8),
|
||||
AutoAcceptCertificates = true,
|
||||
BrowseRoot = "i=85",
|
||||
MaxDiscoveredNodes = 3000,
|
||||
@@ -133,7 +130,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests
|
||||
result.Timeout.ShouldBe(TimeSpan.FromSeconds(12));
|
||||
result.SessionTimeout.ShouldBe(TimeSpan.FromSeconds(150));
|
||||
result.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(7));
|
||||
result.ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(8));
|
||||
result.AutoAcceptCertificates.ShouldBeTrue();
|
||||
result.BrowseRoot.ShouldBe("i=85");
|
||||
result.MaxDiscoveredNodes.ShouldBe(3000);
|
||||
|
||||
+17
@@ -13,6 +13,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as <c>DriverInstanceActor</c>
|
||||
/// runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no
|
||||
/// in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever).
|
||||
/// Since #676 a gracefully stopped server is a DEFINITIVE loss (<c>BadServerHalted</c>), so the driver also raises
|
||||
/// <c>ConnectionLost</c> and the host runs that sequence at once instead of after the watchdog's silence window.
|
||||
/// </summary>
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class OpcUaClientDriverReconnectWireTests
|
||||
@@ -39,6 +41,8 @@ public sealed class OpcUaClientDriverReconnectWireTests
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
|
||||
|
||||
await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-reconnect");
|
||||
var lost = new List<string>();
|
||||
drv.ConnectionLost += (_, e) => { lock (lost) lost.Add(e.Reason); };
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
|
||||
var seen = new List<(uint Status, object? Value)>();
|
||||
@@ -67,12 +71,20 @@ public sealed class OpcUaClientDriverReconnectWireTests
|
||||
frozen.ShouldNotBeNull();
|
||||
await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals
|
||||
drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen);
|
||||
|
||||
// #676: a server that HALTED (graceful stop) is a definitive loss — measured here, the SDK reports
|
||||
// BadServerHalted on every keep-alive and never re-opens the channel, so the session can never recover in
|
||||
// place. The driver says so at once, so the host re-initialises it instead of waiting out the ~90 s watchdog.
|
||||
await WaitAsync(() => { lock (lost) return lost.Count > 0; },
|
||||
"a halted upstream must raise ConnectionLost", TimeSpan.FromSeconds(10));
|
||||
lock (lost) lost.Single().ShouldContain("BadServerHalted");
|
||||
});
|
||||
|
||||
// The restarted server knows nothing of the old session, so the driver stays down on its own …
|
||||
await Task.Delay(TimeSpan.FromSeconds(3), ct);
|
||||
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "nothing in the driver may paper over a dead session");
|
||||
lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive");
|
||||
lock (lost) lost.Count.ShouldBe(1, "once per session — the restarted server's continued refusals raise nothing more");
|
||||
|
||||
// … until the host's comms-loss watchdog re-initialises it (DriverInstanceActor: Reconnecting → ShutdownAsync +
|
||||
// InitializeAsync → Connected → re-subscribe). The new server is empty: redeploy the nodes first, holding the
|
||||
@@ -100,6 +112,7 @@ public sealed class OpcUaClientDriverReconnectWireTests
|
||||
var back = await drv.ReadAsync([OpcUaClientDriverWireTests.SpeedRef], ct);
|
||||
back[0].StatusCode.ShouldBe(0u);
|
||||
Convert.ToSingle(back[0].Value).ShouldBe(88.0f);
|
||||
lock (lost) lost.Count.ShouldBe(1, "the healthy re-initialised session reports no loss");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -117,6 +130,8 @@ public sealed class OpcUaClientDriverReconnectWireTests
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 5.0f);
|
||||
|
||||
await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-sublost");
|
||||
var lost = new List<string>();
|
||||
drv.ConnectionLost += (_, e) => { lock (lost) lost.Add(e.Reason); };
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
var seen = new List<(uint Status, object? Value)>();
|
||||
drv.OnDataChange += (_, e) =>
|
||||
@@ -140,6 +155,8 @@ public sealed class OpcUaClientDriverReconnectWireTests
|
||||
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "… and must not revive a driver whose data plane is dead");
|
||||
drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen);
|
||||
drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running, "the session itself is fine");
|
||||
// #676: a subscription the server dropped never comes back on this session — a definitive loss.
|
||||
lock (lost) lost.ShouldHaveSingleItem().ShouldContain("subscription lost");
|
||||
|
||||
// The watchdog's re-initialise recreates the subscription and the value comes back Good.
|
||||
await drv.ShutdownAsync(ct);
|
||||
|
||||
+144
@@ -0,0 +1,144 @@
|
||||
using Opc.Ua;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #676 item 1 — the IN-PLACE blip (#670): the upstream stops answering but the TCP connection and the
|
||||
/// session survive, keep-alives fail, then traffic resumes before the session times out. Until now only the live
|
||||
/// gate (<c>docker pause</c> of opc-plc) covered it. <see cref="TcpStallProxy"/> reproduces it on loopback: the driver
|
||||
/// connects THROUGH the proxy (which works only because of the #676 endpoint rewrite — the server advertises its own
|
||||
/// port, which would bypass the proxy), and pausing the proxy stalls every byte without closing anything.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Writing this test found a real defect: a change the server published right after the stall ended rode a Publish
|
||||
/// response that had been in flight across the stall; the SDK timed that request out, dropped the response and never
|
||||
/// republished it, so the node stayed Good on the PRE-change value indefinitely. The driver's trailing reconcile read
|
||||
/// (<c>ReconcileAfterRecoveryAsync</c>) is the fix; the "40.0 arrives" leg below goes red without it.
|
||||
/// </remarks>
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class OpcUaClientDriverStallWireTests
|
||||
{
|
||||
private const uint BadNoCommunication = 0x80310000u;
|
||||
|
||||
private static async Task WaitAsync(Func<bool> condition, string because, TimeSpan? timeout = null)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(30));
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (condition()) return;
|
||||
await Task.Delay(25);
|
||||
}
|
||||
condition().ShouldBeTrue(because);
|
||||
}
|
||||
|
||||
private sealed record Seen(uint Status, object? Value, DateTime? Source);
|
||||
|
||||
[Fact]
|
||||
public async Task A_stalled_connection_goes_Bad_once_and_recovers_in_place_on_the_same_session_without_regressing()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-stall", ct);
|
||||
await using var proxy = TcpStallProxy.Start(new Uri(rig.EndpointUrl).Port);
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
|
||||
|
||||
// A 3 s operation timeout keeps the stall short: a keep-alive is declared failed when its request times out.
|
||||
await using var drv = OpcUaClientDriverWireTests.NewDriver(
|
||||
rig, "wire-stall", endpointUrl: proxy.Through(rig.EndpointUrl), timeout: TimeSpan.FromSeconds(3));
|
||||
var connectionLost = 0;
|
||||
drv.ConnectionLost += (_, _) => Interlocked.Increment(ref connectionLost);
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
proxy.AcceptedConnections.ShouldBeGreaterThan(0, "the session must run THROUGH the proxy, or pausing it proves nothing");
|
||||
|
||||
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 connectionsBefore = proxy.AcceptedConnections;
|
||||
var badKeepAlives = 0;
|
||||
session.KeepAlive += (_, e) =>
|
||||
{
|
||||
if (ServiceResult.IsBad(e.Status)) Interlocked.Increment(ref badKeepAlives);
|
||||
};
|
||||
// The SDK's late backlog flush: a data-carrying NotificationMessage whose sequence number is BELOW one already
|
||||
// seen (keep-alive messages carry the next number). It is the positive control for the no-regression check.
|
||||
uint maxSequence = 0;
|
||||
var lateMessages = 0;
|
||||
session.Notification += (_, e) =>
|
||||
{
|
||||
var msg = e.NotificationMessage;
|
||||
if (msg is null) return;
|
||||
lock (seen)
|
||||
{
|
||||
if (msg.SequenceNumber < maxSequence && msg.NotificationData.Count > 0) lateMessages++;
|
||||
maxSequence = Math.Max(maxSequence, msg.SequenceNumber);
|
||||
}
|
||||
};
|
||||
|
||||
// ── Stall ────────────────────────────────────────────────────────────────────────────────────────────────
|
||||
proxy.Pause();
|
||||
await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); },
|
||||
"a stalled upstream must publish BadNoCommunication for the subscribed ref");
|
||||
lock (seen) seen.Single(s => s.Status == BadNoCommunication).Value.ShouldBe(10.0f, "the Bad retains the value");
|
||||
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy);
|
||||
var frozen = drv.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
|
||||
|
||||
// The server's value moves on while the client cannot see it: away and back to the pre-outage value, so the
|
||||
// recovered value is IDENTICAL to the one published Bad — the (value, status) dedup trap.
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 20.0f);
|
||||
await Task.Delay(300, ct); // > one publishing interval, so the server queues 20.0 as its own notification
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
|
||||
|
||||
// Stays Bad and frozen across a further failed keep-alive (counted, not slept for).
|
||||
var badAtDetection = Volatile.Read(ref badKeepAlives);
|
||||
await WaitAsync(() => Volatile.Read(ref badKeepAlives) >= badAtDetection + 2, "further keep-alives must fail while stalled");
|
||||
drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen, "the read stamp is the #527 watchdog's silence measure");
|
||||
lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive");
|
||||
|
||||
// ── Resume well inside the 30 s session timeout ──────────────────────────────────────────────────────────
|
||||
proxy.Resume();
|
||||
await WaitAsync(() => drv.GetHealth().State == DriverState.Healthy, "a resumed stall must recover in place");
|
||||
int badIndex;
|
||||
lock (seen) badIndex = seen.FindIndex(s => s.Status == BadNoCommunication);
|
||||
await WaitAsync(() => { lock (seen) return seen.Skip(badIndex + 1).Any(s => s is { Status: 0u, Value: 10.0f }); },
|
||||
"the identical value must be republished Good after the outage");
|
||||
|
||||
drv.Session.ShouldBeSameAs(session, "an in-place blip must not cost the session");
|
||||
proxy.AcceptedConnections.ShouldBe(connectionsBefore, "and not the TCP connection either");
|
||||
|
||||
// A change published straight after recovery — while the Publish requests held across the stall are still
|
||||
// being answered or timed out — must still reach the node.
|
||||
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 40.0f);
|
||||
await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 40.0f }); },
|
||||
"a change published right after recovery must reach the node (via the subscription or the trailing re-read)");
|
||||
await drv.ReconcileTaskForTest;
|
||||
|
||||
// The SDK flushes the notifications held during the stall seconds after recovery (the #670 F2 shape). Wait for
|
||||
// that flush before judging ordering, or the check below would pass on a run where nothing late arrived.
|
||||
await WaitAsync(() => { lock (seen) return lateMessages > 0; },
|
||||
"the SDK's late backlog flush did not arrive — the no-regression check below would be vacuous");
|
||||
|
||||
lock (seen)
|
||||
{
|
||||
var good = seen.Where(s => s.Status == 0u && s.Source is not null).ToList();
|
||||
for (var i = 1; i < good.Count; i++)
|
||||
good[i].Source!.Value.ShouldBeGreaterThanOrEqualTo(good[i - 1].Source!.Value,
|
||||
$"published Good value #{i} ({good[i].Value}) regressed in source time — a late replay reached the " +
|
||||
$"node. Sequence: [{string.Join(", ", good.Select(s => $"{s.Value}@{s.Source:HH:mm:ss.fff}"))}]");
|
||||
seen.Last().Value.ShouldBe(40.0f, "nothing older may land after the latest value");
|
||||
seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1);
|
||||
}
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
|
||||
drv.Session.ShouldBeSameAs(session);
|
||||
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");
|
||||
}
|
||||
}
|
||||
+3
-4
@@ -41,10 +41,10 @@ public sealed class OpcUaClientDriverWireTests
|
||||
|
||||
internal static OpcUaClientDriver NewDriver(
|
||||
ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList<RawTagEntry>? extraTags = null,
|
||||
TimeSpan? reconnectPeriod = null) =>
|
||||
string? endpointUrl = null, TimeSpan? timeout = null) =>
|
||||
new(new OpcUaClientDriverOptions
|
||||
{
|
||||
EndpointUrl = rig.EndpointUrl,
|
||||
EndpointUrl = endpointUrl ?? rig.EndpointUrl,
|
||||
RawTags =
|
||||
[
|
||||
Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)),
|
||||
@@ -58,11 +58,10 @@ public sealed class OpcUaClientDriverWireTests
|
||||
Username = "apikey",
|
||||
Password = ServerEventRig.KeyPassword,
|
||||
AutoAcceptCertificates = true,
|
||||
Timeout = TimeSpan.FromSeconds(10),
|
||||
Timeout = timeout ?? TimeSpan.FromSeconds(10),
|
||||
SessionTimeout = TimeSpan.FromSeconds(30),
|
||||
KeepAliveInterval = TimeSpan.FromSeconds(1),
|
||||
BrowseRoot = browseRoot,
|
||||
ReconnectPeriod = reconnectPeriod ?? TimeSpan.FromSeconds(5),
|
||||
}, id);
|
||||
|
||||
internal static async Task<ServerEventRig> StartRigAsync(string name, CancellationToken ct)
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #676 — an in-test TCP forwarder that can STALL a connection without closing it: the shape of a
|
||||
/// <c>docker pause</c>, a hung switch port or a congested WAN link, where the socket stays open and simply stops
|
||||
/// carrying bytes. Each accepted connection gets a fresh upstream connection and two copy pumps; every pump awaits a
|
||||
/// shared gate before each write, so <see cref="Pause"/> holds all traffic in both directions (bytes already read
|
||||
/// are held, the rest back up in the kernel buffers) and <see cref="Resume"/> lets it all through in order. Nothing
|
||||
/// is ever closed by a pause — the outage is invisible at the TCP level, exactly as in production.
|
||||
/// </summary>
|
||||
internal sealed class TcpStallProxy : IAsyncDisposable
|
||||
{
|
||||
private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
|
||||
private readonly int _targetPort;
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private readonly List<TcpClient> _sockets = [];
|
||||
private readonly Lock _lock = new();
|
||||
private TaskCompletionSource _gate = Open();
|
||||
private Task _acceptLoop = Task.CompletedTask;
|
||||
|
||||
private TcpStallProxy(int targetPort) => _targetPort = targetPort;
|
||||
|
||||
/// <summary>The loopback port clients connect to.</summary>
|
||||
public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
|
||||
|
||||
/// <summary>Number of client connections accepted so far — a new one means the client re-connected.</summary>
|
||||
public int AcceptedConnections { get; private set; }
|
||||
|
||||
/// <summary>Starts forwarding <c>127.0.0.1:{Port}</c> to <c>127.0.0.1:{targetPort}</c>.</summary>
|
||||
public static TcpStallProxy Start(int targetPort)
|
||||
{
|
||||
var proxy = new TcpStallProxy(targetPort);
|
||||
proxy._listener.Start();
|
||||
proxy._acceptLoop = proxy.AcceptLoopAsync();
|
||||
return proxy;
|
||||
}
|
||||
|
||||
/// <summary>Rewrites the port of an <c>opc.tcp</c> URL to this proxy's.</summary>
|
||||
public string Through(string endpointUrl) => new UriBuilder(endpointUrl) { Port = Port }.Uri.ToString();
|
||||
|
||||
/// <summary>Stops all traffic in both directions; sockets stay open.</summary>
|
||||
public void Pause()
|
||||
{
|
||||
lock (_lock)
|
||||
if (_gate.Task.IsCompleted) _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
}
|
||||
|
||||
/// <summary>Releases everything held since <see cref="Pause"/>, in order.</summary>
|
||||
public void Resume()
|
||||
{
|
||||
lock (_lock) _gate.TrySetResult();
|
||||
}
|
||||
|
||||
private static TaskCompletionSource Open()
|
||||
{
|
||||
var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
tcs.SetResult();
|
||||
return tcs;
|
||||
}
|
||||
|
||||
private Task CurrentGate()
|
||||
{
|
||||
lock (_lock) return _gate.Task;
|
||||
}
|
||||
|
||||
private async Task AcceptLoopAsync()
|
||||
{
|
||||
while (!_cts.IsCancellationRequested)
|
||||
{
|
||||
TcpClient client;
|
||||
try { client = await _listener.AcceptTcpClientAsync(_cts.Token); }
|
||||
catch { return; }
|
||||
var upstream = new TcpClient();
|
||||
try { await upstream.ConnectAsync(IPAddress.Loopback, _targetPort, _cts.Token); }
|
||||
catch
|
||||
{
|
||||
client.Dispose();
|
||||
upstream.Dispose();
|
||||
continue;
|
||||
}
|
||||
lock (_lock)
|
||||
{
|
||||
AcceptedConnections++;
|
||||
_sockets.Add(client);
|
||||
_sockets.Add(upstream);
|
||||
}
|
||||
_ = PumpAsync(client.GetStream(), upstream.GetStream());
|
||||
_ = PumpAsync(upstream.GetStream(), client.GetStream());
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PumpAsync(NetworkStream from, NetworkStream to)
|
||||
{
|
||||
var buffer = new byte[64 * 1024];
|
||||
try
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
var n = await from.ReadAsync(buffer, _cts.Token);
|
||||
if (n == 0) break;
|
||||
await CurrentGate().WaitAsync(_cts.Token);
|
||||
await to.WriteAsync(buffer.AsMemory(0, n), _cts.Token);
|
||||
}
|
||||
}
|
||||
catch { /* either side gone or the proxy disposed */ }
|
||||
// Half-close propagation is not needed: a real close (server stopped) tears the pair down on the next read.
|
||||
try { to.Socket.Shutdown(SocketShutdown.Send); } catch { }
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Resume();
|
||||
await _cts.CancelAsync();
|
||||
_listener.Stop();
|
||||
try { await _acceptLoop; } catch { }
|
||||
lock (_lock)
|
||||
{
|
||||
foreach (var s in _sockets) s.Dispose();
|
||||
_sockets.Clear();
|
||||
}
|
||||
_cts.Dispose();
|
||||
}
|
||||
}
|
||||
+157
@@ -0,0 +1,157 @@
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #676 — a driver that KNOWS its connection is gone (<see cref="IDriverConnectionLossSource"/>) gets
|
||||
/// re-initialised at once instead of after the #527 watchdog's silence window (~90 s in production). The watchdog is
|
||||
/// configured at an hour here, so anything that reconnects within seconds can only be the new signal.
|
||||
/// </summary>
|
||||
public sealed class DriverInstanceActorConnectionLossSignalTests : RuntimeActorTestBase
|
||||
{
|
||||
private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50);
|
||||
|
||||
/// <summary>A permanently Healthy (fresh-read) stub — the watchdog can never fire for it — that can raise
|
||||
/// <see cref="IDriverConnectionLossSource.ConnectionLost"/> on demand and counts its attached handlers.</summary>
|
||||
private sealed class LossSourceStubDriver : StubDriver, IDriverConnectionLossSource
|
||||
{
|
||||
private EventHandler<DriverConnectionLossEventArgs>? _handlers;
|
||||
private int _handlerCount;
|
||||
|
||||
private int _healthReads;
|
||||
|
||||
public int HandlerCount => Volatile.Read(ref _handlerCount);
|
||||
|
||||
/// <summary>Health-poll observations — the positive control for an absence assertion.</summary>
|
||||
public int HealthReadCount => Volatile.Read(ref _healthReads);
|
||||
|
||||
public override DriverHealth GetHealth()
|
||||
{
|
||||
Interlocked.Increment(ref _healthReads);
|
||||
return base.GetHealth();
|
||||
}
|
||||
|
||||
public event EventHandler<DriverConnectionLossEventArgs>? ConnectionLost
|
||||
{
|
||||
add { _handlers += value; Interlocked.Increment(ref _handlerCount); }
|
||||
remove { _handlers -= value; Interlocked.Decrement(ref _handlerCount); }
|
||||
}
|
||||
|
||||
/// <summary>Raised from a pool thread, as a real driver's keep-alive callback would.</summary>
|
||||
public Task RaiseAsync(string reason) =>
|
||||
Task.Run(() => _handlers?.Invoke(this, new DriverConnectionLossEventArgs(reason)));
|
||||
}
|
||||
|
||||
private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(LossSourceStubDriver driver)
|
||||
{
|
||||
var parent = CreateTestProbe();
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
|
||||
driver,
|
||||
reconnectInterval: TimeSpan.FromHours(1),
|
||||
healthPollInterval: FastPoll,
|
||||
commsLossTimeout: TimeSpan.FromHours(1)));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue();
|
||||
return (actor, parent);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_connection_loss_from_a_connected_driver_reconnects_at_once_not_after_the_watchdog()
|
||||
{
|
||||
var driver = new LossSourceStubDriver();
|
||||
var (_, parent) = Connect(driver);
|
||||
var started = DateTime.UtcNow;
|
||||
|
||||
await driver.RaiseAsync("keep-alive reported BadServerHalted");
|
||||
|
||||
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeFalse();
|
||||
(DateTime.UtcNow - started).ShouldBeLessThan(TimeSpan.FromSeconds(3),
|
||||
"the watchdog is configured at an hour, so only the connection-loss signal can have done this");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The falsification control: the same driver and actor with the event suppressed stays Connected — so the
|
||||
/// Reconnecting above is the signal's doing, not something the setup would have done anyway.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Without_a_raise_the_same_driver_stays_connected()
|
||||
{
|
||||
var driver = new LossSourceStubDriver();
|
||||
var (_, parent) = Connect(driver);
|
||||
|
||||
var ticksAtConnect = driver.HealthReadCount;
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(600));
|
||||
(driver.HealthReadCount - ticksAtConnect).ShouldBeGreaterThanOrEqualTo(5,
|
||||
"health-poll ticks did not run — the absence assertion above would be vacuous");
|
||||
driver.HandlerCount.ShouldBe(1, "attached, so the absence above is not merely an unwired event");
|
||||
driver.InitializeCount.ShouldBe(1, "and nothing re-initialised the driver");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A raise while NOT Connected is ignored, not deferred: here it lands while the initial connect is still in
|
||||
/// flight. Acting on it later would tear down the connection that very attempt is building.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_raise_while_connecting_is_ignored()
|
||||
{
|
||||
var driver = new LossSourceStubDriver();
|
||||
var release = new TaskCompletionSource();
|
||||
var initStarted = new TaskCompletionSource();
|
||||
driver.InitBehavior = async _ =>
|
||||
{
|
||||
initStarted.TrySetResult();
|
||||
await release.Task;
|
||||
};
|
||||
var parent = CreateTestProbe();
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
|
||||
driver,
|
||||
reconnectInterval: TimeSpan.FromHours(1),
|
||||
healthPollInterval: FastPoll,
|
||||
commsLossTimeout: TimeSpan.FromHours(1)));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
await initStarted.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
||||
|
||||
// The raise is enqueued before the connect can complete (RaiseAsync returns after the Tell), so the mailbox
|
||||
// delivers it while the actor is still Connecting.
|
||||
await driver.RaiseAsync("a loss reported mid-connect");
|
||||
release.SetResult();
|
||||
|
||||
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue();
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(600));
|
||||
driver.InitializeCount.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_handler_is_attached_once_and_detached_when_the_actor_stops()
|
||||
{
|
||||
var driver = new LossSourceStubDriver();
|
||||
var (actor, parent) = Connect(driver);
|
||||
driver.HandlerCount.ShouldBe(1);
|
||||
|
||||
parent.Watch(actor);
|
||||
Sys.Stop(actor);
|
||||
// PostStop announces comms loss (03/S15) before the Terminated arrives.
|
||||
parent.FishForMessage(m => m is Terminated t && t.ActorRef.Equals(actor), TimeSpan.FromSeconds(3));
|
||||
|
||||
driver.HandlerCount.ShouldBe(0, "the IDriver outlives the actor on a respawn — a leaked handler holds a dead Self");
|
||||
}
|
||||
|
||||
/// <summary>A driver that does not implement the interface is simply never subscribed — the signal is opt-in.</summary>
|
||||
[Fact]
|
||||
public void A_driver_without_the_capability_connects_as_before()
|
||||
{
|
||||
var driver = new StubDriver();
|
||||
var parent = CreateTestProbe();
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
|
||||
driver, reconnectInterval: TimeSpan.FromHours(1), healthPollInterval: FastPoll,
|
||||
commsLossTimeout: TimeSpan.FromHours(1)));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue();
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user