test: add E2E leaf node tests (hub-to-leaf, leaf-to-hub, subject propagation)

Fix LeafNodeManager.StartAsync to launch solicited connections for RemoteLeaves
(config-file-parsed remotes) in addition to the programmatic Remotes list, and
update ParseEndpoint to handle nats-leaf:// scheme URLs. Add LeafNodeFixture
that polls /leafz for connection readiness and three E2E tests covering
hub→leaf delivery, leaf→hub delivery, and subject-scoped propagation.
This commit is contained in:
Joseph Doherty
2026-03-12 19:47:40 -04:00
parent 338f44b07b
commit aeb60d3c43
2 changed files with 108 additions and 0 deletions

View File

@@ -239,6 +239,16 @@ public sealed class LeafNodeManager : IAsyncDisposable
foreach (var remote in _options.Remotes.Distinct(StringComparer.OrdinalIgnoreCase))
_ = Task.Run(() => ConnectSolicitedWithRetryAsync(remote, _options.JetStreamDomain, _cts.Token));
// Also start solicited connections for remotes parsed from the config file (RemoteLeaves).
// RemoteLeaves are populated by the config parser from leafnodes.remotes[] blocks;
// _options.Remotes is the simple programmatic list only.
// Go reference: leafnode.go — createLeafNode starts solicited connections via connectToRemoteLeaf.
foreach (var remoteLeaf in _options.RemoteLeaves)
{
foreach (var url in remoteLeaf.Urls.Distinct(StringComparer.OrdinalIgnoreCase))
_ = Task.Run(() => ConnectSolicitedWithRetryAsync(url, _options.JetStreamDomain, _cts.Token));
}
_logger.LogDebug("Leaf manager started (listen={Host}:{Port})", _options.Host, _options.Port);
return Task.CompletedTask;
}
@@ -819,6 +829,12 @@ public sealed class LeafNodeManager : IAsyncDisposable
private static IPEndPoint ParseEndpoint(string endpoint)
{
// Handle full URLs with a scheme (e.g. "nats-leaf://127.0.0.1:5222").
// Uri.TryCreate handles both schemed URLs and bare "host:port" strings.
if (Uri.TryCreate(endpoint, UriKind.Absolute, out var uri))
return new IPEndPoint(IPAddress.Parse(uri.Host), uri.Port);
// Fall back to bare "host:port" splitting for plain strings without a scheme.
var parts = endpoint.Split(':', 2, StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries);
if (parts.Length != 2)
throw new FormatException($"Invalid endpoint: {endpoint}");