Skip to content

Retry telemetry receiver test setup when a probed port is taken #146

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Views2k merged 5 commits into from
Oct 21, 2026
Merged

Retry telemetry receiver test setup when a probed port is taken #146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
84b2f9c
Retry telemetry receiver test setup when a probed port is taken
Views2k Oct 21, 2026
6d7b40b
Merge remote-tracking branch 'origin/main' into fix/telemetry-test-po…
Views2k Oct 21, 2026
3a72f75
Default the excluded port for the receiver start helper
Views2k Oct 21, 2026
e366e1a
Keep the original released-port probes
Views2k Oct 21, 2026
0629bb5
Merge remote-tracking branch 'origin/main' into fix/telemetry-test-po…
Views2k Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 20 additions & 4 deletions tests/Wisp.Telemetry.Tests/TelemetryPacketDiagnosticsTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -176,6 +174,24 @@ private static byte[] PacketForCar(int car)
return packet;
}

private static async Task<int> StartOnAvailablePortAsync(TelemetryUdpReceiver receiver)
{
// Probing releases the port before the receiver binds it. Another test's
// probe or listener can bind 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)
Expand Down
62 changes: 37 additions & 25 deletions tests/Wisp.Telemetry.Tests/TelemetryUdpReceiverTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,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)!;
Expand All @@ -19,7 +19,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();
Expand All @@ -44,10 +44,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);
Expand All @@ -70,11 +68,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);
Expand All @@ -96,17 +93,15 @@ 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);
Expand All @@ -128,9 +123,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);
Expand Down Expand Up @@ -162,10 +156,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);
Expand All @@ -183,9 +176,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++)
Expand All @@ -206,9 +198,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);
Expand All @@ -229,9 +220,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);
Expand All @@ -245,9 +235,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.
Expand All @@ -274,9 +263,8 @@ 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();

Expand All @@ -290,6 +278,30 @@ public async Task DisposeReleasesPortAndPreventsRestart()
await Assert.ThrowsAsync<ObjectDisposedException>(() => receiver.RestartAsync(GetAvailablePortExcept(port), TestContext.Current.CancellationToken));
}

private static Task<int> StartOnAvailablePortAsync(TelemetryUdpReceiver receiver) =>
BindAvailablePortAsync(port => receiver.StartAsync(port, TestContext.Current.CancellationToken));

private static Task<int> RestartOnAvailablePortAsync(TelemetryUdpReceiver receiver, int excludedPort = 0) =>
BindAvailablePortAsync(port => receiver.RestartAsync(port, TestContext.Current.CancellationToken), excludedPort);

private static async Task<int> BindAvailablePortAsync(Func<int, Task> bind, int excludedPort = 0)
{
// Probing releases the port before the receiver binds it. Another test's
// probe or listener can bind 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 int GetAvailablePort()
{
using var reservation = CreateExclusiveReservation();
Expand Down
Loading