From 84b2f9c971bd79e083da1d405a518ff42d0ece4b Mon Sep 17 00:00:00 2001 From: Views2k <292889387+Views2k@users.noreply.github.com> Date: Wed, 21 Oct 2026 00:56:39 +0000 Subject: [PATCH 1/3] Retry telemetry receiver test setup when a probed port is taken The receiver and packet diagnostics tests probe a free UDP port, close the probe, then start the receiver on that port. A parallel test can bind the port in between, so the start fails with AddressAlreadyInUse even though nothing is wrong with the receiver. Start and restart now pick a fresh port and retry on that error only, matching the existing helpers in the capture and observer tests. The occupied-port cases still bind the port they reserve. The released-port checks accept a refused probe when the port is no longer bound on loopback, since another test can take a released port on the wildcard address. --- .../TelemetryPacketDiagnosticsTests.cs | 24 ++++- .../TelemetryUdpReceiverTests.cs | 102 +++++++++++------- 2 files changed, 83 insertions(+), 43 deletions(-) diff --git a/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs b/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs index 0189976b..723448ab 100644 --- a/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs +++ b/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs @@ -27,8 +27,7 @@ public async Task DisabledRecordingAndThrowingObserversPreserveParsing(int obser }; } receiver.PacketAvailable += (_, _) => accepted.Release(); - var port = GetAvailablePort(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); var endpoint = new IPEndPoint(IPAddress.Loopback, port); var started = Stopwatch.GetTimestamp(); @@ -101,8 +100,7 @@ public async Task DrainedDatagramsKeepTheirOwnIdentityAndNewestParseTimestamp(bo throw new InvalidOperationException("Drained observer failure"); }; receiver.PacketAvailable += (_, _) => accepted.Release(); - var port = GetAvailablePort(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); var endpoint = new IPEndPoint(IPAddress.Loopback, port); try @@ -176,6 +174,24 @@ private static byte[] PacketForCar(int car) return packet; } + private static async Task StartOnAvailablePortAsync(TelemetryUdpReceiver receiver) + { + // Probing releases the port before the receiver binds it. Another test's + // ephemeral sender can claim it in between; retry only that setup race. + for (var attempt = 0; ; attempt++) + { + var port = GetAvailablePort(); + try + { + await receiver.StartAsync(port, TestContext.Current.CancellationToken); + return port; + } + catch (SocketException error) when (error.SocketErrorCode == SocketError.AddressAlreadyInUse && attempt < 7) + { + } + } + } + private static int GetAvailablePort() { using var reservation = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) diff --git a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs index 83005b24..291e97d9 100644 --- a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs +++ b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs @@ -1,4 +1,5 @@ using System.Net; +using System.Net.NetworkInformation; using System.Net.Sockets; using Xunit; @@ -10,7 +11,7 @@ public sealed class TelemetryUdpReceiverTests public async Task UnexpectedSocketClosureRetainsFaultCodeAndRestartClearsIt() { await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(GetAvailablePort(), TestContext.Current.CancellationToken); + await StartOnAvailablePortAsync(receiver); var session = typeof(TelemetryUdpReceiver).GetField("_session", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(receiver)!; var socket = (Socket)session.GetType().GetProperty("Socket")!.GetValue(session)!; var receive = (Task)session.GetType().GetProperty("ReceiveTask")!.GetValue(session)!; @@ -19,7 +20,7 @@ public async Task UnexpectedSocketClosureRetainsFaultCodeAndRestartClearsIt() Assert.NotNull(receiver.GetStatistics(DateTimeOffset.UtcNow).ListenerError); Assert.NotNull(receiver.ListenerFailure); Assert.NotEqual(0, receiver.ListenerFailure.Code); - await receiver.RestartAsync(GetAvailablePort(), TestContext.Current.CancellationToken); + await RestartOnAvailablePortAsync(receiver); Assert.Null(receiver.ListenerFailure); Assert.Null(receiver.GetStatistics(DateTimeOffset.UtcNow).ListenerError); await receiver.StopAsync(); @@ -44,10 +45,8 @@ public void AcceptsDefaultPort() [Fact] public async Task ReceivesAndParsesLocalhostDatagramEndToEnd() { - var port = GetAvailablePort(); - await using var receiver = new TelemetryUdpReceiver { CapturePublicationTimings = true }; - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); var packet = Fh6PacketFixture.Create(); await sender.SendAsync(packet, new IPEndPoint(IPAddress.Loopback, port), TestContext.Current.CancellationToken); @@ -70,11 +69,10 @@ public async Task ReceivesAndParsesLocalhostDatagramEndToEnd() [Fact] public async Task FailedRestartKeepsExistingListenerAndSettingsStateAlive() { - var originalPort = GetAvailablePort(); using var reservation = CreateExclusiveReservation(); var occupiedPort = ((IPEndPoint)reservation.LocalEndPoint!).Port; await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(originalPort, TestContext.Current.CancellationToken); + var originalPort = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); await sender.SendAsync(Fh6PacketFixture.Create(), new IPEndPoint(IPAddress.Loopback, originalPort), TestContext.Current.CancellationToken); await WaitForCarAsync(receiver, 2468); @@ -96,28 +94,22 @@ public async Task FailedRestartKeepsExistingListenerAndSettingsStateAlive() [Fact] public async Task SuccessfulRestartClearsSessionStateAndReleasesOldPort() { - var originalPort = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(originalPort, TestContext.Current.CancellationToken); + var originalPort = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); await sender.SendAsync(Fh6PacketFixture.Create(), new IPEndPoint(IPAddress.Loopback, originalPort), TestContext.Current.CancellationToken); await WaitForCarAsync(receiver, 2468); // Probe after the sender has auto-bound, so its ephemeral source port // cannot claim our released replacement port before the restart. - var replacementPort = GetAvailablePortExcept(originalPort); - await receiver.RestartAsync(replacementPort, TestContext.Current.CancellationToken); + var replacementPort = await RestartOnAvailablePortAsync(receiver, originalPort); Assert.Null(receiver.Latest); Assert.Equal(replacementPort, receiver.ListeningPort); var statistics = receiver.GetStatistics(DateTimeOffset.UtcNow); Assert.Equal(0, statistics.AcceptedPackets); Assert.Equal(0, statistics.RejectedPackets); - using var oldPortProbe = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) - { - ExclusiveAddressUse = true - }; - oldPortProbe.Bind(new IPEndPoint(IPAddress.Loopback, originalPort)); + AssertLoopbackPortReleased(originalPort); var replacementPacket = Fh6PacketFixture.Create(); Fh6PacketFixture.WriteInt32(replacementPacket, 212, 9753); @@ -128,9 +120,8 @@ public async Task SuccessfulRestartClearsSessionStateAndReleasesOldPort() [Fact] public async Task ConcurrentStopsAreIdempotentAndClearLatestState() { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); await sender.SendAsync(Fh6PacketFixture.Create(), new IPEndPoint(IPAddress.Loopback, port), TestContext.Current.CancellationToken); await WaitForCarAsync(receiver, 2468); @@ -162,10 +153,9 @@ public async Task FailedInitialBindDoesNotLeaveReceiverInRunningState() [Fact] public async Task PacketSubscriberFailureDoesNotTerminateListener() { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); receiver.PacketAvailable += (_, _) => throw new InvalidOperationException("Test subscriber failure"); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); await sender.SendAsync(Fh6PacketFixture.Create(), new IPEndPoint(IPAddress.Loopback, port), TestContext.Current.CancellationToken); @@ -183,9 +173,8 @@ public async Task PacketSubscriberFailureDoesNotTerminateListener() [Fact] public async Task PacketBurstRetainsNewestDatagram() { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); for (var index = 0; index < 16; index++) @@ -206,9 +195,8 @@ public async Task PacketBurstRetainsNewestDatagram() [Fact] public async Task MalformedTrafficIsCountedWithoutPretendingValidTelemetryIsFresh() { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); await sender.SendAsync(new byte[8], new IPEndPoint(IPAddress.Loopback, port), TestContext.Current.CancellationToken); @@ -229,9 +217,8 @@ public async Task MalformedTrafficIsCountedWithoutPretendingValidTelemetryIsFres [InlineData(4096)] public async Task OversizedDatagramIsRejectedWithoutStoppingTheListener(int size) { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); var endpoint = new IPEndPoint(IPAddress.Loopback, port); await sender.SendAsync(new byte[size], endpoint, TestContext.Current.CancellationToken); @@ -245,9 +232,8 @@ public async Task OversizedDatagramIsRejectedWithoutStoppingTheListener(int size [Fact] public async Task OversizedDatagramInABurstLeavesTheNewestValidPacket() { - var port = GetAvailablePort(); await using var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); using var sender = new UdpClient(AddressFamily.InterNetwork); var endpoint = new IPEndPoint(IPAddress.Loopback, port); // One burst fits the socket buffer; the listener may drain the later datagrams. @@ -274,22 +260,57 @@ private static async Task WaitForStatistics(TelemetryUdpReceiver receiver, Func< [Fact] public async Task DisposeReleasesPortAndPreventsRestart() { - var port = GetAvailablePort(); var receiver = new TelemetryUdpReceiver(); - await receiver.StartAsync(port, TestContext.Current.CancellationToken); + var port = await StartOnAvailablePortAsync(receiver); await receiver.DisposeAsync(); Assert.False(receiver.IsRunning); Assert.Null(receiver.ListeningPort); - using var portProbe = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) - { - ExclusiveAddressUse = true - }; - portProbe.Bind(new IPEndPoint(IPAddress.Loopback, port)); + AssertLoopbackPortReleased(port); await Assert.ThrowsAsync(() => receiver.RestartAsync(GetAvailablePortExcept(port), TestContext.Current.CancellationToken)); } + private static Task StartOnAvailablePortAsync(TelemetryUdpReceiver receiver) => + BindAvailablePortAsync(port => receiver.StartAsync(port, TestContext.Current.CancellationToken)); + + private static Task RestartOnAvailablePortAsync(TelemetryUdpReceiver receiver, int excludedPort = 0) => + BindAvailablePortAsync(port => receiver.RestartAsync(port, TestContext.Current.CancellationToken), excludedPort); + + private static async Task BindAvailablePortAsync(Func bind, int excludedPort) + { + // Probing releases the port before the receiver binds it. Another test's + // ephemeral sender can claim it in between; retry only that setup race. + for (var attempt = 0; ; attempt++) + { + var port = GetAvailablePortExcept(excludedPort); + try + { + await bind(port); + return port; + } + catch (SocketException error) when (error.SocketErrorCode == SocketError.AddressAlreadyInUse && attempt < 7) + { + } + } + } + + private static void AssertLoopbackPortReleased(int port) + { + try + { + using var probe = CreateExclusiveSocket(); + probe.Bind(new IPEndPoint(IPAddress.Loopback, port)); + } + catch (SocketException error) when (error.SocketErrorCode == SocketError.AddressAlreadyInUse) + { + // Once released, another test's ephemeral sender can take the port on the + // wildcard address. Only a loopback binding means the receiver still holds it. + var loopback = new IPEndPoint(IPAddress.Loopback, port); + Assert.DoesNotContain(loopback, IPGlobalProperties.GetIPGlobalProperties().GetActiveUdpListeners()); + } + } + private static int GetAvailablePort() { using var reservation = CreateExclusiveReservation(); @@ -310,14 +331,17 @@ private static int GetAvailablePortExcept(int excludedPort) private static Socket CreateExclusiveReservation() { - var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) - { - ExclusiveAddressUse = true - }; + var socket = CreateExclusiveSocket(); socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); return socket; } + private static Socket CreateExclusiveSocket() => + new(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) + { + ExclusiveAddressUse = true + }; + private static async Task WaitForCarAsync(TelemetryUdpReceiver receiver, int carOrdinal) { var deadline = DateTimeOffset.UtcNow.AddSeconds(2); From 3a72f75041a306dfa46cb08fccc4e5065e9ef0eb Mon Sep 17 00:00:00 2001 From: Views2k <292889387+Views2k@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:09:28 +0000 Subject: [PATCH 2/3] Default the excluded port for the receiver start helper --- tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs index 291e97d9..632e5754 100644 --- a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs +++ b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs @@ -277,7 +277,7 @@ private static Task StartOnAvailablePortAsync(TelemetryUdpReceiver receiver private static Task RestartOnAvailablePortAsync(TelemetryUdpReceiver receiver, int excludedPort = 0) => BindAvailablePortAsync(port => receiver.RestartAsync(port, TestContext.Current.CancellationToken), excludedPort); - private static async Task BindAvailablePortAsync(Func bind, int excludedPort) + private static async Task BindAvailablePortAsync(Func bind, int excludedPort = 0) { // Probing releases the port before the receiver binds it. Another test's // ephemeral sender can claim it in between; retry only that setup race. From e366e1ac71803e88d7772e1e3a397ca11679cf72 Mon Sep 17 00:00:00 2001 From: Views2k <292889387+Views2k@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:11:27 +0000 Subject: [PATCH 3/3] Keep the original released-port probes --- .../TelemetryPacketDiagnosticsTests.cs | 2 +- .../TelemetryUdpReceiverTests.cs | 42 +++++++------------ 2 files changed, 16 insertions(+), 28 deletions(-) diff --git a/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs b/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs index 723448ab..2a7cf956 100644 --- a/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs +++ b/tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs @@ -177,7 +177,7 @@ private static byte[] PacketForCar(int car) private static async Task StartOnAvailablePortAsync(TelemetryUdpReceiver receiver) { // Probing releases the port before the receiver binds it. Another test's - // ephemeral sender can claim it in between; retry only that setup race. + // probe or listener can bind it in between; retry only that setup race. for (var attempt = 0; ; attempt++) { var port = GetAvailablePort(); diff --git a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs index 632e5754..e4cbe7d5 100644 --- a/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs +++ b/tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs @@ -1,5 +1,4 @@ using System.Net; -using System.Net.NetworkInformation; using System.Net.Sockets; using Xunit; @@ -109,7 +108,11 @@ public async Task SuccessfulRestartClearsSessionStateAndReleasesOldPort() var statistics = receiver.GetStatistics(DateTimeOffset.UtcNow); Assert.Equal(0, statistics.AcceptedPackets); Assert.Equal(0, statistics.RejectedPackets); - AssertLoopbackPortReleased(originalPort); + using var oldPortProbe = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) + { + ExclusiveAddressUse = true + }; + oldPortProbe.Bind(new IPEndPoint(IPAddress.Loopback, originalPort)); var replacementPacket = Fh6PacketFixture.Create(); Fh6PacketFixture.WriteInt32(replacementPacket, 212, 9753); @@ -267,7 +270,11 @@ public async Task DisposeReleasesPortAndPreventsRestart() Assert.False(receiver.IsRunning); Assert.Null(receiver.ListeningPort); - AssertLoopbackPortReleased(port); + using var portProbe = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) + { + ExclusiveAddressUse = true + }; + portProbe.Bind(new IPEndPoint(IPAddress.Loopback, port)); await Assert.ThrowsAsync(() => receiver.RestartAsync(GetAvailablePortExcept(port), TestContext.Current.CancellationToken)); } @@ -280,7 +287,7 @@ private static Task RestartOnAvailablePortAsync(TelemetryUdpReceiver receiv private static async Task BindAvailablePortAsync(Func bind, int excludedPort = 0) { // Probing releases the port before the receiver binds it. Another test's - // ephemeral sender can claim it in between; retry only that setup race. + // probe or listener can bind it in between; retry only that setup race. for (var attempt = 0; ; attempt++) { var port = GetAvailablePortExcept(excludedPort); @@ -295,22 +302,6 @@ private static async Task BindAvailablePortAsync(Func bind, int } } - private static void AssertLoopbackPortReleased(int port) - { - try - { - using var probe = CreateExclusiveSocket(); - probe.Bind(new IPEndPoint(IPAddress.Loopback, port)); - } - catch (SocketException error) when (error.SocketErrorCode == SocketError.AddressAlreadyInUse) - { - // Once released, another test's ephemeral sender can take the port on the - // wildcard address. Only a loopback binding means the receiver still holds it. - var loopback = new IPEndPoint(IPAddress.Loopback, port); - Assert.DoesNotContain(loopback, IPGlobalProperties.GetIPGlobalProperties().GetActiveUdpListeners()); - } - } - private static int GetAvailablePort() { using var reservation = CreateExclusiveReservation(); @@ -331,16 +322,13 @@ private static int GetAvailablePortExcept(int excludedPort) private static Socket CreateExclusiveReservation() { - var socket = CreateExclusiveSocket(); - socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); - return socket; - } - - private static Socket CreateExclusiveSocket() => - new(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) + var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp) { ExclusiveAddressUse = true }; + socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + return socket; + } private static async Task WaitForCarAsync(TelemetryUdpReceiver receiver, int carOrdinal) {