diff --git a/src/CommunityToolkit.Aspire.Hosting.Floci/FlociServiceBusEndpointAllocator.cs b/src/CommunityToolkit.Aspire.Hosting.Floci/FlociServiceBusEndpointAllocator.cs index 399956e41..0991af23f 100644 --- a/src/CommunityToolkit.Aspire.Hosting.Floci/FlociServiceBusEndpointAllocator.cs +++ b/src/CommunityToolkit.Aspire.Hosting.Floci/FlociServiceBusEndpointAllocator.cs @@ -9,6 +9,13 @@ namespace CommunityToolkit.Aspire.Hosting.Floci; internal sealed class FlociServiceBusEndpointAllocator : IDistributedApplicationEventingSubscriber { + // Random host ports are taken from below every OS ephemeral range (Linux starts at 32768, + // Windows and macOS at 49152). An OS-assigned port is released before Docker binds it, and DCP + // can hand it to a project proxy in the meantime: on Windows the proxy's loopback listener then + // shadows Docker's wildcard bind, and on Linux Docker's bind fails. + private const int MinSelectedPort = 20000; + private const int MaxSelectedPortExclusive = 32768; + public Task SubscribeAsync( IDistributedApplicationEventing eventing, DistributedApplicationExecutionContext executionContext, @@ -30,6 +37,7 @@ private static async Task AllocateEndpointsAsync( // DCP does not allocate endpoints on these custom resources. Allocate before the // parent's environment is evaluated, otherwise resolving its child's ports blocks startup. var listeners = new List(); + var reservedPorts = GetReservedPorts(@event.Model); var allocatedResources = new List(); try { @@ -52,7 +60,7 @@ private static async Task AllocateEndpointsAsync( continue; } - endpoint.Port ??= SelectPort(listeners); + endpoint.Port ??= SelectPort(listeners, reservedPorts); endpoint.TargetPort = endpoint.Port; endpoint.AllocatedEndpoint = new AllocatedEndpoint(endpoint, KnownHostNames.Localhost, endpoint.Port.Value, EndpointBindingMode.SingleAddress, null, KnownNetworkIdentifiers.LocalhostNetwork); @@ -86,13 +94,72 @@ await eventing.PublishAsync( } } - private static int SelectPort(List listeners) + internal static HashSet GetReservedPorts(DistributedApplicationModel model) + { + // Ports configured anywhere in the app model are bound later, so probing cannot see them. + // Both reads are synchronous: Port is the configured value and AllocatedEndpoint is null until + // allocation completes, so endpoints that are still being allocated never block startup. + return model.Resources + .SelectMany(resource => resource.Annotations.OfType()) + .SelectMany(endpoint => new[] { endpoint.Port, endpoint.AllocatedEndpoint?.Port }) + .OfType() + .ToHashSet(); + } + + internal static int SelectPort(List listeners, HashSet reservedPorts) + { + // Start at a random offset so concurrent AppHosts spread out, then scan the whole range so a + // free port is always found when one exists. Reserved ports are skipped without probing. + int rangeSize = MaxSelectedPortExclusive - MinSelectedPort; + int offset = Random.Shared.Next(rangeSize); + for (int i = 0; i < rangeSize; i++) + { + int port = MinSelectedPort + (offset + i) % rangeSize; + if (reservedPorts.Contains(port)) + { + continue; + } + + // Match Docker's bind address, as in the k3s allocator, and also probe IPv4 and IPv6 loopback, where + // another process's listener would shadow Docker's wildcard bind. This probes availability; + // it cannot reserve the port through Docker startup, which needs the socket released. + if (!IsAvailable(IPAddress.Loopback, port) + || (Socket.OSSupportsIPv6 && !IsAvailable(IPAddress.IPv6Loopback, port))) + { + continue; + } + + var listener = new TcpListener(IPAddress.Any, port); + try + { + listener.Start(); + } + catch (SocketException) + { + listener.Dispose(); + continue; + } + + listeners.Add(listener); + reservedPorts.Add(port); + return port; + } + + throw new DistributedApplicationException( + $"No free Service Bus host port was found in {MinSelectedPort}-{MaxSelectedPortExclusive - 1}."); + } + + private static bool IsAvailable(IPAddress address, int port) { - // Match Docker's bind address, as in the k3s allocator. This probes availability; - // it cannot reserve the port through Docker startup, which needs the socket released. - var listener = new TcpListener(IPAddress.Any, 0); - listeners.Add(listener); - listener.Start(); - return ((IPEndPoint)listener.LocalEndpoint).Port; + using var listener = new TcpListener(address, port); + try + { + listener.Start(); + return true; + } + catch (SocketException) + { + return false; + } } } diff --git a/tests/CommunityToolkit.Aspire.Hosting.Floci.Tests/AzureServiceBusResourceTests.cs b/tests/CommunityToolkit.Aspire.Hosting.Floci.Tests/AzureServiceBusResourceTests.cs index 90196287d..00c083d02 100644 --- a/tests/CommunityToolkit.Aspire.Hosting.Floci.Tests/AzureServiceBusResourceTests.cs +++ b/tests/CommunityToolkit.Aspire.Hosting.Floci.Tests/AzureServiceBusResourceTests.cs @@ -95,8 +95,6 @@ await builder.Eventing.PublishAsync( .GetValueAsync(TestContext.Current.CancellationToken); string? amqpTlsPort = await serviceBus.Resource.AmqpTlsEndpoint.Property(EndpointProperty.Port) .GetValueAsync(TestContext.Current.CancellationToken); - Assert.InRange(int.Parse(amqpPort!), 1, 65535); - Assert.InRange(int.Parse(amqpTlsPort!), 1, 65535); Assert.NotEqual(amqpPort, amqpTlsPort); foreach (var endpoint in new[] { serviceBus.Resource.AmqpEndpoint, serviceBus.Resource.AmqpTlsEndpoint }) { @@ -110,10 +108,19 @@ await builder.Eventing.PublishAsync( { Assert.Equal(configuredAmqpPort.Value.ToString(), amqpPort); } + else + { + // Selected ports stay below every OS ephemeral range, where DCP allocates proxy ports. + Assert.InRange(int.Parse(amqpPort!), 20000, 32767); + } if (configuredAmqpTlsPort.HasValue) { Assert.Equal(configuredAmqpTlsPort.Value.ToString(), amqpTlsPort); } + else + { + Assert.InRange(int.Parse(amqpTlsPort!), 20000, 32767); + } var resource = Assert.Single(appModel.Resources.OfType()); Assert.True(resource.TryGetAnnotationsOfType(out IEnumerable? envAnnotations)); @@ -137,6 +144,90 @@ await builder.Eventing.PublishAsync( .GetValueAsync(TestContext.Current.CancellationToken)); } + [Fact] + public void GetReservedPortsIncludesConfiguredAndAllocatedPortsOnly() + { + IDistributedApplicationBuilder builder = DistributedApplication.CreateBuilder(); + + builder.AddContainer("configured", "image").WithEndpoint("tcp", endpoint => endpoint.Port = 21000); + builder.AddContainer("allocated", "image").WithEndpoint("tcp", endpoint => + endpoint.AllocatedEndpoint = new AllocatedEndpoint(endpoint, "localhost", 22000)); + builder.AddContainer("pending", "image").WithEndpoint(targetPort: 8080, name: "tcp"); + + using var app = builder.Build(); + var appModel = app.Services.GetRequiredService(); + + Assert.Equal([21000, 22000], FlociServiceBusEndpointAllocator.GetReservedPorts(appModel).Order()); + } + + [Fact] + public void SelectPortSkipsReservedPorts() + { + // Leave exactly one free, unreserved port in the selection range. + int freePort = Enumerable.Range(20000, 12768).First(port => + { + using var probe = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Any, port); + try + { + probe.Start(); + return true; + } + catch (System.Net.Sockets.SocketException) + { + return false; + } + }); + var reserved = Enumerable.Range(20000, 12768).Where(port => port != freePort).ToHashSet(); + var listeners = new List(); + try + { + Assert.Equal(freePort, FlociServiceBusEndpointAllocator.SelectPort(listeners, reserved)); + Assert.Contains(freePort, reserved); + } + finally + { + listeners.ForEach(listener => listener.Stop()); + } + } + + [Fact] + public void SelectPortSkipsPortsHeldOnIPv6Loopback() + { + Assert.SkipUnless(System.Net.Sockets.Socket.OSSupportsIPv6, "IPv6 is not available."); + + // Hold one port on ::1 only, and leave it plus one other free port unreserved. + var free = Enumerable.Range(20000, 12768).Where(IsFreeEverywhere).Take(2).ToArray(); + using var ipv6Holder = new System.Net.Sockets.TcpListener(System.Net.IPAddress.IPv6Loopback, free[0]); + ipv6Holder.Start(); + var reserved = Enumerable.Range(20000, 12768).Where(port => !free.Contains(port)).ToHashSet(); + var listeners = new List(); + try + { + Assert.Equal(free[1], FlociServiceBusEndpointAllocator.SelectPort(listeners, reserved)); + } + finally + { + listeners.ForEach(listener => listener.Stop()); + } + } + + private static bool IsFreeEverywhere(int port) + { + foreach (var address in new[] { System.Net.IPAddress.Any, System.Net.IPAddress.IPv6Loopback }) + { + using var probe = new System.Net.Sockets.TcpListener(address, port); + try + { + probe.Start(); + } + catch (System.Net.Sockets.SocketException) + { + return false; + } + } + return true; + } + [Fact] public async Task WithServiceBusDoesNotConfigureTheDataPlaneInPublishMode() {