diff --git a/src/EventStore.ClusterNode/Components/Pages/Cluster.razor b/src/EventStore.ClusterNode/Components/Pages/Cluster.razor
index 47d3c0279e..e72be103d8 100644
--- a/src/EventStore.ClusterNode/Components/Pages/Cluster.razor
+++ b/src/EventStore.ClusterNode/Components/Pages/Cluster.razor
@@ -43,8 +43,7 @@
Status |
Timestamp (UTC) |
Checkpoints |
- TCP |
- HTTP |
+ HTTP / gRPC |
Actions |
@@ -52,7 +51,7 @@
@if (!ClusterMembers.Any())
{
- | @ClusterEmptyMessage |
+ @ClusterEmptyMessage |
}
else
@@ -77,10 +76,6 @@
@EpochLabel(member)
}
-
- Internal: @InternalTcpEndpoint(member)
- External: @ExternalTcpEndpoint(member)
- |
@HttpEndpoint(member) |
@@ -307,7 +302,7 @@
@@ -370,9 +365,7 @@
.Append("Snapshot taken at ")
.Append(TimestampLabel(ClusterReadAt ?? DateTime.UtcNow))
.AppendLine()
- .Append(PadRight("Internal Tcp", 31)).Append(' ')
- .Append(PadRight("External Tcp", 31)).Append(' ')
- .Append(PadRight("Http", 23)).Append(' ')
+ .Append(PadRight("HTTP / gRPC", 23)).Append(' ')
.Append(PadRight("Status", 11)).Append(' ')
.Append(PadRight("State", 18)).Append(' ')
.Append(PadRight("Timestamp (UTC)", 19)).Append(" Checkpoints");
@@ -380,8 +373,6 @@
foreach (var member in ClusterMembers)
{
builder.AppendLine()
- .Append(PadRight(InternalTcpEndpoint(member), 31)).Append(' ')
- .Append(PadRight(ExternalTcpEndpoint(member), 31)).Append(' ')
.Append(PadRight(HttpEndpoint(member), 23)).Append(' ')
.Append(PadRight(MemberStatus(member), 11)).Append(' ')
.Append(PadRight(member.State.ToString(), 18)).Append(' ')
@@ -500,16 +491,6 @@
private static string MemberStatus(ClientClusterInfo.ClientMemberInfo member) =>
member.IsAlive ? "Alive" : "Unreachable";
- private static string InternalTcpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
- Endpoint(
- member.InternalTcpIp,
- member.InternalSecureTcpPort != 0 ? member.InternalSecureTcpPort : member.InternalTcpPort);
-
- private static string ExternalTcpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
- Endpoint(
- member.ExternalTcpIp,
- member.ExternalSecureTcpPort != 0 ? member.ExternalSecureTcpPort : member.ExternalTcpPort);
-
private static string HttpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
Endpoint(member.HttpEndPointIp, member.HttpEndPointPort);
diff --git a/src/EventStore.ClusterNode/Components/Services/ClusterStatusService.cs b/src/EventStore.ClusterNode/Components/Services/ClusterStatusService.cs
index 5c753987ec..78ca5f48ac 100644
--- a/src/EventStore.ClusterNode/Components/Services/ClusterStatusService.cs
+++ b/src/EventStore.ClusterNode/Components/Services/ClusterStatusService.cs
@@ -171,7 +171,7 @@ private ClusterReplicaRow ParseReplicaRow(
: Guid.Empty;
var totalBytesSent = row.TotalBytesSent;
var previousRow = _previousReplicas.GetValueOrDefault(connectionId);
- var replicaNode = FindMemberByInternalEndpoint(members, row.SubscriptionEndpoint);
+ var replicaNode = FindMemberByEndpoint(members, row.SubscriptionEndpoint);
var isCatchingUp = replicaNode?.State == VNodeState.CatchingUp;
var catchupStartTime = now;
var catchupStartBytesSent = totalBytesSent;
@@ -206,12 +206,12 @@ private ClusterReplicaRow ParseReplicaRow(
private ClaimsPrincipal CurrentUser =>
httpContextAccessor.HttpContext?.User ?? new ClaimsPrincipal(new ClaimsIdentity());
- private static ClientClusterInfo.ClientMemberInfo FindMemberByInternalEndpoint(
+ private static ClientClusterInfo.ClientMemberInfo FindMemberByEndpoint(
IReadOnlyList members,
string endpoint)
{
var cleaned = endpoint.Replace("Unspecified/", "", StringComparison.OrdinalIgnoreCase);
- return members.FirstOrDefault(x => string.Equals(InternalTcpEndpoint(x), cleaned, StringComparison.OrdinalIgnoreCase));
+ return members.FirstOrDefault(x => string.Equals(HttpEndpoint(x), cleaned, StringComparison.OrdinalIgnoreCase));
}
private static Uri BuildLeaderAddress(
@@ -219,11 +219,6 @@ private static Uri BuildLeaderAddress(
ClientClusterInfo.ClientMemberInfo leader) =>
new UriBuilder(request.Scheme, leader.HttpEndPointIp, leader.HttpEndPointPort).Uri;
- private static string InternalTcpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
- Endpoint(
- member.InternalTcpIp,
- member.InternalSecureTcpPort != 0 ? member.InternalSecureTcpPort : member.InternalTcpPort);
-
private static string HttpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
Endpoint(member.HttpEndPointIp, member.HttpEndPointPort);
diff --git a/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs b/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs
new file mode 100644
index 0000000000..8b98cc7262
--- /dev/null
+++ b/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs
@@ -0,0 +1,278 @@
+using System;
+using System.Buffers;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+
+namespace EventStore.ClusterNode.Components.Services;
+
+public sealed class NodeConnectionTracker
+{
+ private readonly ConcurrentDictionary _connections = new();
+
+ public IReadOnlyList Snapshot() =>
+ _connections.Values.Select(x => x.Snapshot())
+ .OrderBy(x => x.RemoteEndPoint, StringComparer.OrdinalIgnoreCase)
+ .ThenBy(x => x.ConnectionId, StringComparer.Ordinal)
+ .ToArray();
+
+ public async Task Track(ConnectionContext context, ConnectionDelegate next, bool isTls)
+ {
+ var state = new NodeConnectionState(
+ context.ConnectionId,
+ context.RemoteEndPoint?.ToString() ?? "",
+ context.LocalEndPoint?.ToString() ?? "",
+ isTls,
+ DateTimeOffset.UtcNow);
+ _connections[context.ConnectionId] = state;
+ context.Transport = new CountingDuplexPipe(context.Transport, state);
+
+ try
+ {
+ await next(context);
+ }
+ finally
+ {
+ _connections.TryRemove(context.ConnectionId, out _);
+ }
+ }
+
+ public void ObserveRequest(
+ string connectionId,
+ string protocol,
+ bool isGrpc,
+ string connectionName,
+ string userAgent)
+ {
+ if (_connections.TryGetValue(connectionId, out var connection))
+ connection.ObserveRequest(protocol, isGrpc, connectionName, userAgent);
+ }
+}
+
+public sealed record NodeConnectionSnapshot(
+ string ConnectionId,
+ string RemoteEndPoint,
+ string LocalEndPoint,
+ string ClientName,
+ string Application,
+ string Protocol,
+ bool IsTls,
+ DateTimeOffset ConnectedAt,
+ long TotalBytesSent,
+ long TotalBytesReceived,
+ long PendingSendBytes,
+ long PendingReceivedBytes);
+
+internal sealed class NodeConnectionState
+{
+ private readonly object _metadataLock = new();
+ private string _clientName = "";
+ private bool _hasExplicitConnectionName;
+ private bool _hasGrpcRequests;
+ private bool _hasHttpRequests;
+ private long _pendingReceivedBytes;
+ private long _pendingSendBytes;
+ private string _protocol = "";
+ private long _totalBytesReceived;
+ private long _totalBytesSent;
+
+ public NodeConnectionState(
+ string connectionId,
+ string remoteEndPoint,
+ string localEndPoint,
+ bool isTls,
+ DateTimeOffset connectedAt)
+ {
+ ConnectionId = connectionId;
+ RemoteEndPoint = remoteEndPoint;
+ LocalEndPoint = localEndPoint;
+ IsTls = isTls;
+ ConnectedAt = connectedAt;
+ }
+
+ private string ConnectionId { get; }
+ private string RemoteEndPoint { get; }
+ private string LocalEndPoint { get; }
+ private bool IsTls { get; }
+ private DateTimeOffset ConnectedAt { get; }
+
+ public void Received(long bytes, long pendingBytes)
+ {
+ Interlocked.Add(ref _totalBytesReceived, bytes);
+ Interlocked.Exchange(ref _pendingReceivedBytes, pendingBytes);
+ }
+
+ public void Reading(long pendingBytes) =>
+ Interlocked.Exchange(ref _pendingReceivedBytes, pendingBytes);
+
+ public void Sending(int bytes)
+ {
+ Interlocked.Add(ref _totalBytesSent, bytes);
+ Interlocked.Add(ref _pendingSendBytes, bytes);
+ }
+
+ public void Sent() => Interlocked.Exchange(ref _pendingSendBytes, 0);
+
+ public void ObserveRequest(
+ string protocol,
+ bool isGrpc,
+ string connectionName,
+ string userAgent)
+ {
+ lock (_metadataLock)
+ {
+ _protocol = Merge(_protocol, protocol);
+ _hasGrpcRequests |= isGrpc;
+ _hasHttpRequests |= !isGrpc;
+
+ if (!string.IsNullOrWhiteSpace(connectionName))
+ {
+ _clientName = connectionName;
+ _hasExplicitConnectionName = true;
+ }
+ else if (!_hasExplicitConnectionName && !string.IsNullOrWhiteSpace(userAgent))
+ {
+ _clientName = userAgent;
+ }
+ }
+ }
+
+ public NodeConnectionSnapshot Snapshot()
+ {
+ lock (_metadataLock)
+ {
+ return new(
+ ConnectionId,
+ RemoteEndPoint,
+ LocalEndPoint,
+ _clientName,
+ ApplicationLabel(),
+ _protocol,
+ IsTls,
+ ConnectedAt,
+ Interlocked.Read(ref _totalBytesSent),
+ Interlocked.Read(ref _totalBytesReceived),
+ Interlocked.Read(ref _pendingSendBytes),
+ Interlocked.Read(ref _pendingReceivedBytes));
+ }
+ }
+
+ private string ApplicationLabel() => (_hasHttpRequests, _hasGrpcRequests) switch
+ {
+ (true, true) => "HTTP and gRPC",
+ (false, true) => "gRPC",
+ (true, false) => "HTTP",
+ _ => "Awaiting request"
+ };
+
+ private static string Merge(string current, string observed)
+ {
+ if (string.IsNullOrWhiteSpace(observed) || current == observed)
+ return current;
+ return string.IsNullOrWhiteSpace(current) ? observed : "Mixed";
+ }
+}
+
+internal sealed class CountingDuplexPipe : IDuplexPipe
+{
+ public CountingDuplexPipe(IDuplexPipe inner, NodeConnectionState state)
+ {
+ Input = new CountingPipeReader(inner.Input, state);
+ Output = new CountingPipeWriter(inner.Output, state);
+ }
+
+ public PipeReader Input { get; }
+ public PipeWriter Output { get; }
+}
+
+internal sealed class CountingPipeReader : PipeReader
+{
+ private readonly PipeReader _inner;
+ private readonly NodeConnectionState _state;
+ private ReadOnlySequence _currentBuffer;
+
+ public CountingPipeReader(PipeReader inner, NodeConnectionState state)
+ {
+ _inner = inner;
+ _state = state;
+ }
+
+ public override void AdvanceTo(SequencePosition consumed) => AdvanceTo(consumed, consumed);
+
+ public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
+ {
+ var consumedBytes = _currentBuffer.IsEmpty ? 0 : _currentBuffer.Slice(0, consumed).Length;
+ var pendingBytes = _currentBuffer.IsEmpty ? 0 : _currentBuffer.Slice(consumed).Length;
+ _state.Received(consumedBytes, pendingBytes);
+ _currentBuffer = default;
+ _inner.AdvanceTo(consumed, examined);
+ }
+
+ public override void CancelPendingRead() => _inner.CancelPendingRead();
+
+ public override void Complete(Exception exception = null) => _inner.Complete(exception);
+
+ public override ValueTask CompleteAsync(Exception exception = null) => _inner.CompleteAsync(exception);
+
+ public override async ValueTask ReadAsync(CancellationToken cancellationToken = default)
+ {
+ var result = await _inner.ReadAsync(cancellationToken);
+ Observe(result);
+ return result;
+ }
+
+ public override bool TryRead(out ReadResult result)
+ {
+ if (!_inner.TryRead(out result))
+ return false;
+
+ Observe(result);
+ return true;
+ }
+
+ private void Observe(ReadResult result)
+ {
+ _currentBuffer = result.Buffer;
+ _state.Reading(result.Buffer.Length);
+ }
+}
+
+internal sealed class CountingPipeWriter : PipeWriter
+{
+ private readonly PipeWriter _inner;
+ private readonly NodeConnectionState _state;
+
+ public CountingPipeWriter(PipeWriter inner, NodeConnectionState state)
+ {
+ _inner = inner;
+ _state = state;
+ }
+
+ public override void Advance(int bytes)
+ {
+ _state.Sending(bytes);
+ _inner.Advance(bytes);
+ }
+
+ public override void CancelPendingFlush() => _inner.CancelPendingFlush();
+
+ public override void Complete(Exception exception = null) => _inner.Complete(exception);
+
+ public override ValueTask CompleteAsync(Exception exception = null) => _inner.CompleteAsync(exception);
+
+ public override async ValueTask FlushAsync(CancellationToken cancellationToken = default)
+ {
+ var result = await _inner.FlushAsync(cancellationToken);
+ if (!result.IsCanceled)
+ _state.Sent();
+ return result;
+ }
+
+ public override Memory GetMemory(int sizeHint = 0) => _inner.GetMemory(sizeHint);
+
+ public override Span GetSpan(int sizeHint = 0) => _inner.GetSpan(sizeHint);
+}
diff --git a/src/EventStore.ClusterNode/Components/Services/ReplicationEndpointPolicy.cs b/src/EventStore.ClusterNode/Components/Services/ReplicationEndpointPolicy.cs
new file mode 100644
index 0000000000..1c8c2f7fd3
--- /dev/null
+++ b/src/EventStore.ClusterNode/Components/Services/ReplicationEndpointPolicy.cs
@@ -0,0 +1,32 @@
+using System;
+using System.Net;
+using Microsoft.AspNetCore.Http;
+
+namespace EventStore.ClusterNode.Components.Services;
+
+public sealed class ReplicationEndpointPolicy(IPEndPoint listenEndPoint)
+{
+ private const string ReplicationServicePath = "/event_store.replication.Replication";
+
+ public bool Allows(HttpContext context)
+ {
+ var isReplicationConnection = Matches(context.Connection.LocalIpAddress, context.Connection.LocalPort);
+ var isReplicationRequest = context.Request.Path.StartsWithSegments(
+ ReplicationServicePath,
+ StringComparison.Ordinal);
+
+ return isReplicationConnection == isReplicationRequest;
+ }
+
+ private bool Matches(IPAddress localAddress, int localPort)
+ {
+ if (localPort != listenEndPoint.Port)
+ {
+ return false;
+ }
+
+ return listenEndPoint.Address.Equals(IPAddress.Any) ||
+ listenEndPoint.Address.Equals(IPAddress.IPv6Any) ||
+ listenEndPoint.Address.Equals(localAddress);
+ }
+}
diff --git a/src/EventStore.ClusterNode/Program.cs b/src/EventStore.ClusterNode/Program.cs
index 270308e3c5..6dbe3b2dbb 100644
--- a/src/EventStore.ClusterNode/Program.cs
+++ b/src/EventStore.ClusterNode/Program.cs
@@ -25,6 +25,7 @@
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.DataProtection;
using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
@@ -273,6 +274,10 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
x.SuppressStatusMessages = true;
});
+ var nodeConnectionTracker = new NodeConnectionTracker();
+ var replicationEndpointPolicy = new ReplicationEndpointPolicy(
+ new System.Net.IPEndPoint(options.Interface.ReplicationIp, options.Interface.ReplicationPort));
+ builder.Services.AddSingleton(nodeConnectionTracker);
builder.WebHost.ConfigureKestrel(server =>
{
server.Limits.Http2.KeepAlivePingDelay =
@@ -281,12 +286,15 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
TimeSpan.FromMilliseconds(options.Grpc.KeepAliveTimeout);
server.Listen(options.Interface.NodeIp, options.Interface.NodePort, listenOptions =>
- ConfigureHttpOptions(listenOptions, hostedService,
+ ConfigureHttpOptions(listenOptions, hostedService, nodeConnectionTracker,
useHttps: !hostedService.Node.DisableHttps));
+ server.Listen(options.Interface.ReplicationIp, options.Interface.ReplicationPort, listenOptions =>
+ ConfigureHttpOptions(listenOptions, hostedService, nodeConnectionTracker,
+ useHttps: !hostedService.Node.DisableHttps, http2Only: true));
if (hostedService.Node.EnableUnixSocket)
{
- TryListenOnUnixSocket(hostedService, server);
+ TryListenOnUnixSocket(hostedService, server, nodeConnectionTracker);
}
});
@@ -325,6 +333,29 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
builder.Services.AddSingleton(hostedService);
var app = builder.Build();
+ app.Use((context, next) =>
+ {
+ var isGrpc = context.Request.ContentType?.StartsWith(
+ "application/grpc",
+ StringComparison.OrdinalIgnoreCase) == true;
+ nodeConnectionTracker.ObserveRequest(
+ context.Connection.Id,
+ context.Request.Protocol,
+ isGrpc,
+ context.Request.Headers["connection-name"].FirstOrDefault(),
+ context.Request.Headers.UserAgent.ToString());
+ return next(context);
+ });
+ app.Use(async (context, next) =>
+ {
+ if (!replicationEndpointPolicy.Allows(context))
+ {
+ context.Response.StatusCode = StatusCodes.Status404NotFound;
+ return;
+ }
+
+ await next(context);
+ });
app.UseMiddleware();
hostedService.Node.Startup.Configure(app);
if (oauthEnabled)
@@ -376,21 +407,34 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
}
}
- private static void ConfigureHttpOptions(ListenOptions listenOptions, ClusterVNodeHostedService hostedService,
- bool useHttps)
+ private static void ConfigureHttpOptions(
+ ListenOptions listenOptions,
+ ClusterVNodeHostedService hostedService,
+ NodeConnectionTracker connectionTracker,
+ bool useHttps,
+ bool http2Only = false)
{
+ listenOptions.Use(next => context => connectionTracker.Track(context, next, useHttps));
+ if (http2Only)
+ {
+ listenOptions.Protocols = HttpProtocols.Http2;
+ }
+
if (useHttps)
{
listenOptions.UseHttps(CreateServerOptionsSelectionCallback(hostedService), null);
}
- else
+ else if (!http2Only)
{
listenOptions.Use(next =>
new ClearTextHttpMultiplexingMiddleware(next).OnConnectAsync);
}
}
- private static void TryListenOnUnixSocket(ClusterVNodeHostedService hostedService, KestrelServerOptions server)
+ private static void TryListenOnUnixSocket(
+ ClusterVNodeHostedService hostedService,
+ KestrelServerOptions server,
+ NodeConnectionTracker connectionTracker)
{
if (!RuntimeInformation.IsLinux && !OperatingSystem.IsWindowsVersionAtLeast(10, 0, 17063))
{
@@ -421,7 +465,7 @@ private static void TryListenOnUnixSocket(ClusterVNodeHostedService hostedServic
server.ListenUnixSocket(unixSocket, listenOptions =>
{
listenOptions.Use(next => new UnixSocketConnectionMiddleware(next).OnConnectAsync);
- ConfigureHttpOptions(listenOptions, hostedService, useHttps: false);
+ ConfigureHttpOptions(listenOptions, hostedService, connectionTracker, useHttps: false);
});
Log.Information("Listening on UNIX domain socket: {unixSocket}", unixSocket);
}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/EventDataComparer.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/EventDataComparer.cs
deleted file mode 100644
index 15a424be6c..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/EventDataComparer.cs
+++ /dev/null
@@ -1,46 +0,0 @@
-using EventStore.ClientAPI;
-using EventStore.Common.Utils;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-internal static class EventDataComparer
-{
- public static bool Equal(EventData expected, RecordedEvent actual)
- {
- if (expected.EventId != actual.EventId)
- {
- return false;
- }
-
- if (expected.Type != actual.EventType)
- {
- return false;
- }
-
- var expectedDataString = Helper.UTF8NoBom.GetString(expected.Data ?? new byte[0]);
- var expectedMetadataString = Helper.UTF8NoBom.GetString(expected.Metadata ?? new byte[0]);
-
- var actualDataString = Helper.UTF8NoBom.GetString(actual.Data ?? new byte[0]);
- var actualMetadataDataString = Helper.UTF8NoBom.GetString(actual.Metadata ?? new byte[0]);
-
- return expectedDataString == actualDataString && expectedMetadataString == actualMetadataDataString;
- }
-
- public static bool Equal(EventData[] expected, RecordedEvent[] actual)
- {
- if (expected.Length != actual.Length)
- {
- return false;
- }
-
- for (var i = 0; i < expected.Length; i++)
- {
- if (!Equal(expected[i], actual[i]))
- {
- return false;
- }
- }
-
- return true;
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/EventsStream.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/EventsStream.cs
deleted file mode 100644
index da4bb62ce2..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/EventsStream.cs
+++ /dev/null
@@ -1,25 +0,0 @@
-using System.Threading.Tasks;
-using EventStore.ClientAPI;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-internal class EventsStream
-{
- private const int SliceSize = 10;
-
- public static async Task Count(IEventStoreConnection store, string stream)
- {
- var result = 0;
- while (true)
- {
- var slice = await store.ReadStreamEventsForwardAsync(stream, result, SliceSize, false);
- result += slice.Events.Length;
- if (slice.IsEndOfStream)
- {
- break;
- }
- }
-
- return result;
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/TcpType.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/TcpType.cs
deleted file mode 100644
index b860eb63ac..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/TcpType.cs
+++ /dev/null
@@ -1,7 +0,0 @@
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-public enum TcpType
-{
- Normal,
- Ssl
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnection.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnection.cs
deleted file mode 100644
index 674773889c..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnection.cs
+++ /dev/null
@@ -1,72 +0,0 @@
-using System;
-using System.Net;
-using System.Threading;
-using EventStore.ClientAPI;
-using EventStore.ClientAPI.Internal;
-using EventStore.ClientAPI.SystemData;
-using EventStore.Core.Tests.Helpers;
-using NUnit.Framework;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-public static class TestConnection
-{
- private static int _nextConnId = -1;
-
- public static IEventStoreConnection Create(IPEndPoint endPoint, TcpType tcpType = TcpType.Ssl,
- UserCredentials userCredentials = null)
- {
- return EventStoreConnection.Create(Settings(tcpType, userCredentials),
- endPoint.ToESTcpUri(),
- $"ESC-{Interlocked.Increment(ref _nextConnId)}");
- }
-
- public static IEventStoreConnection CreateMiniNodeClient(IPEndPoint endPoint, TcpType tcpType = TcpType.Ssl,
- UserCredentials userCredentials = null)
- {
- return EventStoreConnection.Create(Settings(
- tcpType,
- userCredentials,
- limitAttemptsForOperationTo: 10,
- reconnectionDelay: TimeSpan.FromMilliseconds(100)),
- endPoint.ToESTcpUri(),
- $"ESC-{Interlocked.Increment(ref _nextConnId)}");
- }
-
- public static IEventStoreConnection To(MiniNode miniNode, TcpType tcpType,
- UserCredentials userCredentials = null)
- {
- return EventStoreConnection.Create(Settings(tcpType, userCredentials),
- miniNode.TcpEndPoint.ToESTcpUri(),
- $"ESC-{Interlocked.Increment(ref _nextConnId)}");
- }
-
- private static ConnectionSettingsBuilder Settings(
- TcpType tcpType,
- UserCredentials userCredentials,
- int limitAttemptsForOperationTo = 1,
- TimeSpan? reconnectionDelay = null)
- {
- var settings = ConnectionSettings.Create()
- .SetDefaultUserCredentials(userCredentials)
- .UseCustomLogger(ClientApiLoggerBridge.Default)
- .EnableVerboseLogging()
- .LimitReconnectionsTo(10)
- .LimitAttemptsForOperationTo(limitAttemptsForOperationTo)
- .SetTimeoutCheckPeriodTo(TimeSpan.FromMilliseconds(100))
- .SetReconnectionDelayTo(reconnectionDelay ?? TimeSpan.Zero)
- .FailOnNoServerResponse()
- //.SetOperationTimeoutTo(TimeSpan.FromDays(1))
- ;
- if (tcpType == TcpType.Ssl)
- {
- settings.DisableServerCertificateValidation();
- }
- else
- {
- settings.DisableTls();
- }
-
- return settings;
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnectionLifecycle.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnectionLifecycle.cs
deleted file mode 100644
index 9ea1365f78..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestConnectionLifecycle.cs
+++ /dev/null
@@ -1,77 +0,0 @@
-using System;
-using System.Threading.Tasks;
-using EventStore.ClientAPI;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-public static class TestConnectionLifecycle
-{
- public static async Task ReconnectUntilReady(
- Func createConnection,
- Func readinessProbe,
- TimeSpan timeout)
- {
- var deadline = DateTime.UtcNow + timeout;
-
- while (true)
- {
- IEventStoreConnection connection = null;
-
- try
- {
- connection = createConnection();
- await connection.ConnectAsync();
- await readinessProbe(connection);
- return connection;
- }
- catch (Exception ex)
- {
- if (connection != null)
- {
- TryCloseConnection(connection);
- }
-
- if (IsTransientConnectionFailure(ex) && DateTime.UtcNow < deadline)
- {
- await Task.Delay(250);
- continue;
- }
-
- throw;
- }
- }
- }
-
- public static async Task CloseConnectionAndWait(IEventStoreConnection connection, TimeSpan timeout)
- {
- var closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
- connection.Closed += (_, _) => closed.TrySetResult();
- connection.Close();
- await closed.Task.WithTimeout(timeout);
- }
-
- public static bool IsTransientConnectionFailure(Exception ex) =>
- ex.GetType().Name is "ConnectionClosedException"
- or "RetriesLimitReachedException"
- or "NotAuthenticatedException"
- or "AccessDeniedException";
-
- public static void TryCloseConnection(IEventStoreConnection connection)
- {
- try
- {
- connection.Close();
- }
- catch
- {
- }
- }
-
- public static void DisposeIfNeeded(object candidate)
- {
- if (candidate is IDisposable disposable)
- {
- disposable.Dispose();
- }
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestEvent.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/TestEvent.cs
deleted file mode 100644
index 8fb92bedf3..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/TestEvent.cs
+++ /dev/null
@@ -1,22 +0,0 @@
-using System;
-using System.Text;
-using EventStore.ClientAPI;
-using EventStore.Common.Utils;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-public class TestEvent
-{
- public static EventData NewTestEvent(string data = null, string metadata = null, string eventName = "TestEvent")
- {
- return NewTestEvent(Guid.NewGuid(), data, metadata, eventName);
- }
-
- public static EventData NewTestEvent(Guid eventId, string data = null, string metadata = null, string eventName = "TestEvent")
- {
- var encodedData = Helper.UTF8NoBom.GetBytes(data ?? eventId.ToString());
- var encodedMetadata = Helper.UTF8NoBom.GetBytes(metadata ?? "metadata");
-
- return new EventData(eventId, eventName, false, encodedData, encodedMetadata);
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/Helpers/Writer.cs b/src/EventStore.Core.Tests/ClientAPI/Helpers/Writer.cs
deleted file mode 100644
index 236e47ccb6..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/Helpers/Writer.cs
+++ /dev/null
@@ -1,98 +0,0 @@
-using System.Threading.Tasks;
-using EventStore.ClientAPI;
-using NUnit.Framework;
-
-namespace EventStore.Core.Tests.ClientAPI.Helpers;
-
-internal class StreamWriter
-{
- private readonly IEventStoreConnection _store;
- private readonly string _stream;
- private readonly long _version;
-
- public StreamWriter(IEventStoreConnection store, string stream, long version)
- {
- _store = store;
- _stream = stream;
- _version = version;
- }
-
- public async Task Append(params EventData[] events)
- {
- for (var i = 0; i < events.Length; i++)
- {
- var expVer = _version == ExpectedVersion.Any ? ExpectedVersion.Any : _version + i;
- var nextExpVer = (await _store.AppendToStreamAsync(_stream, expVer, new[] { events[i] })).NextExpectedVersion;
- if (_version != ExpectedVersion.Any)
- {
- Assert.AreEqual(expVer + 1, nextExpVer);
- }
- }
-
- return new TailWriter(_store, _stream);
- }
-}
-
-internal class TailWriter
-{
- private readonly IEventStoreConnection _store;
- private readonly string _stream;
-
- public TailWriter(IEventStoreConnection store, string stream)
- {
- _store = store;
- _stream = stream;
- }
-
- public async Task Then(EventData @event, long expectedVersion)
- {
- await _store.AppendToStreamAsync(_stream, expectedVersion, new[] { @event });
- return this;
- }
-}
-
-internal class TransactionalWriter
-{
- private readonly IEventStoreConnection _store;
- private readonly string _stream;
-
- public TransactionalWriter(IEventStoreConnection store, string stream)
- {
- _store = store;
- _stream = stream;
- }
-
- public async Task StartTransaction(long expectedVersion)
- {
- return new OngoingTransaction(await _store.StartTransactionAsync(_stream, expectedVersion));
- }
-
- public OngoingTransaction ContinueTransaction(long transactionId)
- {
- return new OngoingTransaction(_store.ContinueTransaction(transactionId));
- }
-}
-
-//TODO GFY this should be removed and merged with the public idea of a transaction.
-internal class OngoingTransaction
-{
- private readonly EventStoreTransaction _transaction;
-
- public long TransactionId => _transaction.TransactionId;
-
- public OngoingTransaction(EventStoreTransaction transaction)
- {
- _transaction = transaction;
- }
-
- public async Task Write(params EventData[] events)
- {
- await _transaction.WriteAsync(events);
- return this;
- }
-
- public Task Commit()
- {
- return _transaction.CommitAsync();
- }
-}
diff --git a/src/EventStore.Core.Tests/ClientAPI/SpecificationWithMiniNode.cs b/src/EventStore.Core.Tests/ClientAPI/SpecificationWithMiniNode.cs
deleted file mode 100644
index ddd01c25d0..0000000000
--- a/src/EventStore.Core.Tests/ClientAPI/SpecificationWithMiniNode.cs
+++ /dev/null
@@ -1,102 +0,0 @@
-using System;
-using System.Threading.Tasks;
-using EventStore.ClientAPI;
-using EventStore.Core.Tests.ClientAPI.Helpers;
-using EventStore.Core.Tests.Helpers;
-using NUnit.Framework;
-
-namespace EventStore.Core.Tests.ClientAPI;
-
-public abstract class SpecificationWithMiniNode : SpecificationWithDirectoryPerTestFixture
-{
- private readonly int _chunkSize;
- protected MiniNode _node;
- protected IEventStoreConnection _conn;
- protected virtual TimeSpan Timeout { get; } = TimeSpan.FromMinutes(1);
- protected virtual TimeSpan StartupTimeout => TimeSpan.FromMinutes(5);
-
- protected virtual Task Given() => Task.CompletedTask;
-
- protected abstract Task When();
-
- protected virtual IEventStoreConnection BuildConnection(MiniNode node)
- {
- return TestConnection.CreateMiniNodeClient(node.TcpEndPoint, TcpType.Ssl);
- }
-
- protected Task CloseConnectionAndWait(IEventStoreConnection connection) =>
- TestConnectionLifecycle.CloseConnectionAndWait(connection, Timeout);
-
- protected SpecificationWithMiniNode() : this(chunkSize: 1024 * 1024) { }
-
- protected SpecificationWithMiniNode(int chunkSize)
- {
- _chunkSize = chunkSize;
- }
-
- [OneTimeSetUp]
- public override async Task TestFixtureSetUp()
- {
-
- MiniNodeLogging.Setup();
-
- try
- {
- await base.TestFixtureSetUp();
- }
- catch (Exception ex)
- {
- throw new Exception("TestFixtureSetUp Failed", ex);
- }
-
- try
- {
- _node = new MiniNode(PathName, chunkSize: _chunkSize);
- await _node.Start(StartupTimeout);
- await _node.WaitForTcpEndPoint().WithTimeout(StartupTimeout);
- _conn = await TestConnectionLifecycle.ReconnectUntilReady(
- () => BuildConnection(_node),
- connection => connection.ReadAllEventsForwardAsync(Position.Start, 1, false, DefaultData.AdminCredentials),
- StartupTimeout);
- }
- catch (Exception ex)
- {
- MiniNodeLogging.WriteLogs();
- throw new Exception("MiniNodeSetUp Failed", ex);
- }
-
- try
- {
- await Given().WithTimeout(Timeout);
- }
- catch (Exception ex)
- {
- MiniNodeLogging.WriteLogs();
- throw new Exception("Given Failed", ex);
- }
-
- try
- {
- await When().WithTimeout(Timeout);
- }
- catch (Exception ex)
- {
- MiniNodeLogging.WriteLogs();
- throw new Exception("When Failed", ex);
- }
- }
-
- [OneTimeTearDown]
- public override async Task TestFixtureTearDown()
- {
- if (_conn != null)
- {
- await TestConnectionLifecycle.CloseConnectionAndWait(_conn, Timeout);
- }
-
- await _node.Shutdown();
- await base.TestFixtureTearDown();
-
- MiniNodeLogging.Clear();
- }
-}
diff --git a/src/EventStore.Core.Tests/Cluster/MemberInfoTests.cs b/src/EventStore.Core.Tests/Cluster/MemberInfoTests.cs
index 14d94644ce..b2bf2b68ea 100644
--- a/src/EventStore.Core.Tests/Cluster/MemberInfoTests.cs
+++ b/src/EventStore.Core.Tests/Cluster/MemberInfoTests.cs
@@ -1,5 +1,6 @@
using System;
using System.Net;
+using System.Reflection;
using EventStore.Core.Data;
using NUnit.Framework;
@@ -16,12 +17,7 @@ public void member_with_dns_endpoint_should_equal()
var memberWithDnsEndPoint = EventStore.Core.Cluster.MemberInfo.Initial(Guid.Empty, DateTime.UtcNow,
VNodeState.Unknown, true,
new DnsEndPoint(ipAddress, port),
- new DnsEndPoint(ipAddress, port),
- new DnsEndPoint(ipAddress, port),
- new DnsEndPoint(ipAddress, port),
- new DnsEndPoint(ipAddress, port),
- null, 0, 0,
- 0, false);
+ null, 0, 0, false);
var ipEndPoint = new IPEndPoint(IPAddress.Parse(ipAddress), port);
var dnsEndPoint = new DnsEndPoint(ipAddress, port);
@@ -38,11 +34,7 @@ public void member_with_ip_endpoint_should_equal()
var memberWithDnsEndPoint = EventStore.Core.Cluster.MemberInfo.Initial(Guid.Empty, DateTime.UtcNow,
VNodeState.Unknown, true,
new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- null, 0, 0, 0, false);
+ null, 0, 0, false);
var ipEndPoint = new IPEndPoint(IPAddress.Parse(ipAddress), port);
var dnsEndPoint = new DnsEndPoint(ipAddress, port);
@@ -50,4 +42,61 @@ public void member_with_ip_endpoint_should_equal()
Assert.True(memberWithDnsEndPoint.Is(ipEndPoint));
Assert.True(memberWithDnsEndPoint.Is(dnsEndPoint));
}
+
+ [Test]
+ public void internal_gossip_round_trip_preserves_the_grpc_replication_endpoint()
+ {
+ var replicationEndPoint = new DnsEndPoint("replication-node", 1112);
+ var httpEndPoint = new DnsEndPoint("public-node", 2113);
+ var member = EventStore.Core.Cluster.MemberInfo.Initial(
+ Guid.NewGuid(),
+ DateTime.UtcNow,
+ VNodeState.Unknown,
+ true,
+ httpEndPoint,
+ null,
+ 0,
+ 0,
+ false,
+ replicationEndPoint: replicationEndPoint);
+
+ var grpc = ToGrpcClusterInfo(new EventStore.Core.Cluster.ClusterInfo(member));
+ var roundTrip = FromGrpcClusterInfo(grpc);
+
+ Assert.That(roundTrip.Members, Has.Length.EqualTo(1));
+ Assert.That(roundTrip.Members[0].HttpEndPoint, Is.EqualTo(httpEndPoint));
+ Assert.That(roundTrip.Members[0].ReplicationEndPoint, Is.EqualTo(replicationEndPoint));
+ }
+
+ [Test]
+ public void member_without_a_replication_endpoint_uses_its_http_endpoint()
+ {
+ var httpEndPoint = new DnsEndPoint("mixed-version-node", 2113);
+ var member = EventStore.Core.Cluster.MemberInfo.Initial(
+ Guid.NewGuid(),
+ DateTime.UtcNow,
+ VNodeState.Unknown,
+ true,
+ httpEndPoint,
+ null,
+ 0,
+ 0,
+ false);
+ var grpc = ToGrpcClusterInfo(new EventStore.Core.Cluster.ClusterInfo(member));
+ grpc.Members[0].ReplicationEndPoint = null;
+
+ var roundTrip = FromGrpcClusterInfo(grpc);
+
+ Assert.That(roundTrip.Members[0].ReplicationEndPoint, Is.EqualTo(httpEndPoint));
+ }
+
+ private static EventStore.Cluster.ClusterInfo ToGrpcClusterInfo(EventStore.Core.Cluster.ClusterInfo clusterInfo) =>
+ (EventStore.Cluster.ClusterInfo)typeof(EventStore.Core.Cluster.ClusterInfo)
+ .GetMethod("ToGrpcClusterInfo", BindingFlags.Static | BindingFlags.NonPublic)
+ .Invoke(null, [clusterInfo]);
+
+ private static EventStore.Core.Cluster.ClusterInfo FromGrpcClusterInfo(EventStore.Cluster.ClusterInfo clusterInfo) =>
+ (EventStore.Core.Cluster.ClusterInfo)typeof(EventStore.Core.Cluster.ClusterInfo)
+ .GetMethod("FromGrpcClusterInfo", BindingFlags.Static | BindingFlags.NonPublic)
+ .Invoke(null, [clusterInfo, null]);
}
diff --git a/src/EventStore.Core.Tests/DefaultData.cs b/src/EventStore.Core.Tests/DefaultData.cs
index 3c647b08fc..262f7312dc 100644
--- a/src/EventStore.Core.Tests/DefaultData.cs
+++ b/src/EventStore.Core.Tests/DefaultData.cs
@@ -1,5 +1,4 @@
using System.Net;
-using EventStore.ClientAPI.SystemData;
using EventStore.Core.Services;
namespace EventStore.Core.Tests;
@@ -8,7 +7,6 @@ public class DefaultData
{
public static string AdminUsername = SystemUsers.Admin;
public static string AdminPassword = SystemUsers.DefaultAdminPassword;
- public static UserCredentials AdminCredentials = new UserCredentials(AdminUsername, AdminPassword);
public static NetworkCredential AdminNetworkCredentials = new NetworkCredential(AdminUsername, AdminPassword);
public static ClusterVNodeOptions.DefaultUserOptions DefaultUserOptions = new ClusterVNodeOptions.DefaultUserOptions()
{
diff --git a/src/EventStore.Core.Tests/Helpers/ClientApiLoggerBridge.cs b/src/EventStore.Core.Tests/Helpers/ClientApiLoggerBridge.cs
deleted file mode 100644
index 862b03493f..0000000000
--- a/src/EventStore.Core.Tests/Helpers/ClientApiLoggerBridge.cs
+++ /dev/null
@@ -1,92 +0,0 @@
-using System;
-using EventStore.Common.Utils;
-using ILogger = Serilog.ILogger;
-
-namespace EventStore.Core.Tests.Helpers;
-
-public class ClientApiLoggerBridge : EventStore.ClientAPI.ILogger
-{
- public static readonly ClientApiLoggerBridge Default =
- new ClientApiLoggerBridge(Serilog.Log.ForContext(Serilog.Core.Constants.SourceContextPropertyName,
- "client-api"));
-
- private readonly Serilog.ILogger _log;
-
- public ClientApiLoggerBridge(ILogger log)
- {
- Ensure.NotNull(log, "log");
- _log = log;
- }
-
- public void Error(string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Error(format);
- }
- else
- {
- _log.Error(format, args);
- }
- }
-
- public void Error(Exception ex, string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Error(ex, format);
- }
- else
- {
- _log.Error(ex, format, args);
- }
- }
-
- public void Info(string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Information(format);
- }
- else
- {
- _log.Information(format, args);
- }
- }
-
- public void Info(Exception ex, string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Information(ex, format);
- }
- else
- {
- _log.Information(ex, format, args);
- }
- }
-
- public void Debug(string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Debug(format);
- }
- else
- {
- _log.Debug(format, args);
- }
- }
-
- public void Debug(Exception ex, string format, params object[] args)
- {
- if (args.Length == 0)
- {
- _log.Debug(ex, format);
- }
- else
- {
- _log.Debug(ex, format, args);
- }
- }
-}
diff --git a/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs b/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs
index 444ab592d2..d5b3f61487 100644
--- a/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs
+++ b/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs
@@ -20,11 +20,10 @@
using EventStore.Core.Services.Monitoring;
using EventStore.Core.Services.PersistentSubscription.ConsumerStrategy;
using EventStore.Core.Services.Storage.ReaderIndex;
-using EventStore.Core.Tests.Services.Transport.Tcp;
using EventStore.Core.TransactionLog.Chunks;
using EventStore.Plugins.Subsystems;
-using EventStore.TcpUnitTestPlugin;
using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.AspNetCore.Server.Kestrel.Https;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
@@ -42,9 +41,8 @@ public class MiniClusterNode
private static readonly ILogger Log = Serilog.Log.ForContext>();
- public IPEndPoint InternalTcpEndPoint { get; }
- public IPEndPoint ExternalTcpEndPoint { get; }
public IPEndPoint HttpEndPoint { get; }
+ public IPEndPoint ReplicationEndPoint { get; }
public readonly int DebugIndex;
@@ -62,11 +60,11 @@ public class MiniClusterNode
public VNodeState NodeState = VNodeState.Unknown;
private readonly IHost _host;
- public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp, IPEndPoint externalTcp,
- IPEndPoint httpEndPoint, EndPoint[] gossipSeeds, ISubsystem[] subsystems = null,
+ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint nodeEndPoint, IPEndPoint replicationEndPoint,
+ EndPoint[] gossipSeeds, ISubsystem[] subsystems = null,
bool enableTrustedAuth = false, int memTableSize = 1000,
bool disableFlushToDisk = false, bool readOnlyReplica = false, int nodePriority = 0,
- string intHostAdvertiseAs = null, IExpiryStrategy expiryStrategy = null,
+ IExpiryStrategy expiryStrategy = null,
ArchiveOptions archiveOptions = null, bool archiver = false,
int clusterSize = 3, bool unsafeAllowSurplusNodes = false)
{
@@ -75,19 +73,17 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
RunCount += 1;
DebugIndex = debugIndex;
- InternalTcpEndPoint = internalTcp;
- ExternalTcpEndPoint = externalTcp;
- HttpEndPoint = httpEndPoint;
+ HttpEndPoint = nodeEndPoint;
+ ReplicationEndPoint = replicationEndPoint;
_dbPath = Path.Combine(
pathname,
- $"mini-cluster-node-db-{externalTcp.Port}-{httpEndPoint.Port}");
+ $"mini-cluster-node-db-{nodeEndPoint.Port}");
Directory.CreateDirectory(_dbPath);
FileStreamExtensions.ConfigureFlush(disableFlushToDisk);
subsystems ??= [];
- subsystems = [.. subsystems, new TcpApiTestPlugin()];
var options = new ClusterVNodeOptions
{
@@ -117,14 +113,11 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
},
Interface = new()
{
- ReplicationIp = InternalTcpEndPoint.Address,
- NodeIp = ExternalTcpEndPoint.Address,
- ReplicationPort = InternalTcpEndPoint.Port,
+ NodeIp = HttpEndPoint.Address,
NodePort = HttpEndPoint.Port,
- ReplicationHeartbeatTimeout = 2_000,
- ReplicationHeartbeatInterval = 2_000,
- EnableTrustedAuth = enableTrustedAuth,
- ReplicationHostAdvertiseAs = intHostAdvertiseAs
+ ReplicationIp = ReplicationEndPoint.Address,
+ ReplicationPort = ReplicationEndPoint.Port,
+ EnableTrustedAuth = enableTrustedAuth
},
Database = new()
{
@@ -149,14 +142,7 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
PlugableComponents = subsystems
};
- var configuration = new List> {
- new("EventStore:TcpPlugin:NodeTcpPort", externalTcp.Port.ToString()),
- new("EventStore:TcpPlugin:EnableExternalTcp", "true"),
- new("EventStore:TcpUnitTestPlugin:NodeTcpPort", externalTcp.Port.ToString()),
- new("EventStore:TcpUnitTestPlugin:NodeHeartbeatInterval", "10000"),
- new("EventStore:TcpUnitTestPlugin:NodeHeartbeatTimeout", "10000"),
- new("EventStore:TcpUnitTestPlugin:Insecure", options.Application.Insecure.ToString()),
- };
+ var configuration = new List>();
if (archiveOptions is not null)
{
@@ -176,9 +162,9 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
var inMemConf = new ConfigurationBuilder()
.AddInMemoryCollection(configuration)
.Build();
- var serverCertificate = ssl_connections.GetServerCertificate();
+ var serverCertificate = TestCertificates.GetServerCertificate();
var trustedRootCertificates =
- new X509Certificate2Collection(ssl_connections.GetRootCertificate());
+ new X509Certificate2Collection(TestCertificates.GetRootCertificate());
options = options.Secure(trustedRootCertificates, serverCertificate);
_isReadOnlyReplica = readOnlyReplica;
@@ -191,8 +177,8 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
RuntimeInformation.RuntimeMode, "GC:",
GC.MaxGeneration == 0
? "NON-GENERATION (PROBABLY BOEHM)"
- : $"{GC.MaxGeneration + 1} GENERATIONS", "DBPATH:", _dbPath, "ExTCP ENDPOINT:",
- ExternalTcpEndPoint, "ExHTTP ENDPOINT:", HttpEndPoint);
+ : $"{GC.MaxGeneration + 1} GENERATIONS", "DBPATH:", _dbPath, "NODE ENDPOINT:",
+ HttpEndPoint, "HTTP ENDPOINT:", HttpEndPoint);
var logFormatFactory = LogFormatHelper.LogFormatFactory;
Node = new ClusterVNode(options, logFormatFactory, new AuthenticationProviderFactory(
@@ -215,7 +201,7 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
webHost
.UseKestrel(o =>
{
- o.Listen(HttpEndPoint, options =>
+ void ConfigureHttps(ListenOptions options)
{
options.UseHttps(new HttpsConnectionAdapterOptions
{
@@ -233,7 +219,10 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
return isValid;
}
});
- });
+ }
+
+ o.Listen(HttpEndPoint, ConfigureHttps);
+ o.Listen(ReplicationEndPoint, ConfigureHttps);
})
.UseStartup(Node.Startup);
})
diff --git a/src/EventStore.Core.Tests/Helpers/MiniNode.cs b/src/EventStore.Core.Tests/Helpers/MiniNode.cs
index 9a41dc772b..95dd4d7d5e 100644
--- a/src/EventStore.Core.Tests/Helpers/MiniNode.cs
+++ b/src/EventStore.Core.Tests/Helpers/MiniNode.cs
@@ -5,7 +5,6 @@
using System.Linq;
using System.Net;
using System.Net.Http;
-using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using EventStore.Common.Utils;
@@ -20,13 +19,11 @@
using EventStore.Core.Services.Monitoring;
using EventStore.Core.Services.Storage.ReaderIndex;
using EventStore.Core.Tests.Index.Hashers;
-using EventStore.Core.Tests.Services.Transport.Tcp;
using EventStore.Core.TransactionLog.Chunks;
using EventStore.Plugins.Authentication;
using EventStore.Plugins.Authorization;
using EventStore.Plugins.Subsystems;
using EventStore.Plugins.Transforms;
-using EventStore.TcpUnitTestPlugin;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.AspNetCore.Server.Kestrel.Https;
@@ -45,8 +42,6 @@ public class MiniNode
public const int CachedChunkSize = ChunkSize + ChunkHeader.Size + ChunkFooter.Size;
protected static readonly ILogger Log = Serilog.Log.ForContext();
- public IPEndPoint TcpEndPoint { get; protected set; }
- public IPEndPoint IntTcpEndPoint { get; protected set; }
public IPEndPoint HttpEndPoint { get; protected set; }
}
@@ -71,7 +66,7 @@ public class MiniNode : MiniNode, IAsyncDisposable
public Task AdminUserCreated => _adminUserCreated.Task;
public MiniNode(string pathname,
- int? tcpPort = null, int? httpPort = null,
+ int? httpPort = null,
ISubsystem[] subsystems = null,
int chunkSize = ChunkSize, int cachedChunkSize = CachedChunkSize, bool enableTrustedAuth = false,
int memTableSize = 1000,
@@ -97,26 +92,21 @@ public MiniNode(string pathname,
var ip = IPAddress.Loopback;
- int extTcpPort = tcpPort ?? PortsHelper.GetAvailablePort(ip);
int httpEndPointPort = httpPort ?? PortsHelper.GetAvailablePort(ip);
- int intTcpPort = PortsHelper.GetAvailablePort(ip);
if (string.IsNullOrEmpty(dbPath))
{
DbPath = Path.Combine(pathname,
- $"mini-node-db-{extTcpPort}-{httpEndPointPort}");
+ $"mini-node-db-{httpEndPointPort}");
}
else
{
DbPath = dbPath;
}
- TcpEndPoint = new IPEndPoint(ip, extTcpPort);
- IntTcpEndPoint = new IPEndPoint(ip, intTcpPort);
HttpEndPoint = new IPEndPoint(ip, httpEndPointPort);
subsystems ??= [];
- subsystems = [.. subsystems, new TcpApiTestPlugin()];
var options = new ClusterVNodeOptions
{
@@ -130,8 +120,6 @@ public MiniNode(string pathname,
},
Interface = new()
{
- ReplicationHeartbeatInterval = 10_000,
- ReplicationHeartbeatTimeout = 10_000,
EnableTrustedAuth = enableTrustedAuth
},
Cluster = new()
@@ -162,21 +150,11 @@ public MiniNode(string pathname,
LoadedOptions = ClusterVNodeOptions.GetLoadedOptions(new ConfigurationBuilder()
.AddEventStoreDefaultValues()
.Build()),
- }.Secure(new X509Certificate2Collection(ssl_connections.GetRootCertificate()),
- ssl_connections.GetServerCertificate())
- .WithReplicationEndpointOn(IntTcpEndPoint)
- .WithExternalTcpOn(TcpEndPoint)
+ }.Secure(new X509Certificate2Collection(TestCertificates.GetRootCertificate()),
+ TestCertificates.GetServerCertificate())
.WithNodeEndpointOn(HttpEndPoint);
- var inMemConf = new ConfigurationBuilder()
- .AddInMemoryCollection(new KeyValuePair[] {
- new("EventStore:TcpPlugin:NodeTcpPort", extTcpPort.ToString()),
- new("EventStore:TcpPlugin:EnableExternalTcp", "true"),
- new("EventStore:TcpUnitTestPlugin:NodeTcpPort", extTcpPort.ToString()),
- new("EventStore:TcpUnitTestPlugin:NodeHeartbeatInterval", "10000"),
- new("EventStore:TcpUnitTestPlugin:NodeHeartbeatTimeout", "10000"),
- new("EventStore:TcpUnitTestPlugin:Insecure", options.Application.Insecure.ToString()),
- }).Build();
+ var inMemConf = new ConfigurationBuilder().Build();
if (advertisedExtHostAddress != null)
{
@@ -200,7 +178,7 @@ public MiniNode(string pathname,
? "NON-GENERATION (PROBABLY BOEHM)"
: $"{GC.MaxGeneration + 1} GENERATIONS",
"DBPATH:", DbPath,
- "TCP ENDPOINT:", TcpEndPoint,
+ "NODE ENDPOINT:", HttpEndPoint,
"HTTP ENDPOINT:", HttpEndPoint);
var logFormatFactory = LogFormatHelper.LogFormatFactory
@@ -244,12 +222,12 @@ public MiniNode(string pathname,
{
options.UseHttps(new HttpsConnectionAdapterOptions
{
- ServerCertificate = ssl_connections.GetServerCertificate(),
+ ServerCertificate = TestCertificates.GetServerCertificate(),
ClientCertificateMode = ClientCertificateMode.AllowCertificate,
ClientCertificateValidation = (certificate, chain, sslPolicyErrors) =>
{
var (isValid, error) =
- ClusterVNode.ValidateClientCertificate(certificate, chain, sslPolicyErrors, () => null, () => new X509Certificate2Collection(ssl_connections.GetRootCertificate()));
+ ClusterVNode.ValidateClientCertificate(certificate, chain, sslPolicyErrors, () => null, () => new X509Certificate2Collection(TestCertificates.GetRootCertificate()));
if (!isValid && error != null)
{
Log.Error("Client certificate validation error: {e}", error);
@@ -335,25 +313,6 @@ void WaitForAdminUser(StorageMessage.EventCommitted m)
Log.Information("MiniNode successfully started!");
}
- public async Task WaitForTcpEndPoint()
- {
- while (true)
- {
- using var client = new TcpClient();
-
- try
- {
- await client.ConnectAsync(TcpEndPoint.Address, TcpEndPoint.Port)
- .WaitAsync(TimeSpan.FromMilliseconds(250));
- return;
- }
- catch (Exception ex) when (ex is SocketException or TimeoutException)
- {
- await Task.Delay(100);
- }
- }
- }
-
public async Task Shutdown(bool keepDb = false)
{
diff --git a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs
index 781da7f1ce..d8e41d20c9 100644
--- a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs
+++ b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs
@@ -2,11 +2,12 @@
using System.IO;
using System.Linq;
using System.Net;
+using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Amazon.S3;
using Amazon.S3.Model;
-using EventStore.ClientAPI;
+using EventStore.Client.Streams;
using EventStore.Core.Data;
using EventStore.Core.Messages;
using EventStore.Core.Messaging;
@@ -18,7 +19,11 @@
using EventStore.Core.Tests.Helpers;
using EventStore.Core.TransactionLog.Chunks.TFChunk;
using EventStore.Core.TransactionLog.FileNamingStrategy;
+using Google.Protobuf;
+using Grpc.Net.Client;
using NUnit.Framework;
+using GrpcMetadata = EventStore.Core.Services.Transport.Grpc.Constants.Metadata;
+using StreamsClient = EventStore.Client.Streams.Streams.StreamsClient;
namespace EventStore.Core.Tests.Integration.Archive;
@@ -44,6 +49,8 @@ public class when_archiving_and_restoring_a_cluster
private long _archivedCheckpoint;
private int _restoredNodeIndex;
private int _completedIterations;
+ private GrpcChannel _channel;
+ private StreamsClient _client;
protected override int NodeCount => 4;
protected override TimeSpan GivenTimeout => SoakTimeout;
@@ -117,19 +124,13 @@ protected override MiniClusterNode CreateNode(
new(
PathName,
index,
- endpoints.InternalTcp,
- endpoints.ExternalTcp,
- endpoints.HttpEndPoint,
+ endpoints.NodeEndPoint,
+ endpoints.ReplicationEndPoint,
gossipSeeds,
readOnlyReplica: index == ArchiverNodeIndex,
archiveOptions: _archiveOptions.Enabled ? _archiveOptions : null,
archiver: index == ArchiverNodeIndex);
- protected override IEventStoreConnection CreateConnection() =>
- EventStoreConnection.Create(
- ConnectionSettings.Create().DisableServerCertificateValidation(),
- GetLeader().ExternalTcpEndPoint);
-
protected override async Task Given()
{
var payload = new byte[256 * 1024];
@@ -141,10 +142,30 @@ protected override async Task Given()
var leader = await ReconnectToLeader();
for (var eventNumber = 0; eventNumber < EventsPerIteration; eventNumber++)
{
- await _conn.AppendToStreamAsync(
- Stream,
- EventStore.ClientAPI.ExpectedVersion.Any,
- new EventData(Guid.NewGuid(), "archive-event", isJson: false, payload, Array.Empty()));
+ using var call = _client.Append();
+ await call.RequestStream.WriteAsync(new AppendReq
+ {
+ Options = new()
+ {
+ Any = new(),
+ StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(Stream) }
+ }
+ });
+ await call.RequestStream.WriteAsync(new AppendReq
+ {
+ ProposedMessage = new()
+ {
+ Id = Core.Services.Transport.Grpc.Uuid.NewUuid().ToDto(),
+ Data = ByteString.CopyFrom(payload),
+ CustomMetadata = ByteString.Empty,
+ Metadata = {
+ { GrpcMetadata.Type, "archive-event" },
+ { GrpcMetadata.ContentType, GrpcMetadata.ContentTypes.ApplicationOctetStream }
+ }
+ }
+ });
+ await call.RequestStream.CompleteAsync();
+ await call.ResponseAsync;
}
AssertEx.IsOrBecomesTrue(
@@ -175,11 +196,16 @@ await _conn.AppendToStreamAsync(
private async Task> ReconnectToLeader()
{
var leader = GetLeader();
- _conn?.Close();
- _conn = EventStoreConnection.Create(
- ConnectionSettings.Create().DisableServerCertificateValidation(),
- leader.ExternalTcpEndPoint);
- await _conn.ConnectAsync();
+ _channel?.Dispose();
+ _channel = GrpcChannel.ForAddress(new Uri($"https://{leader.HttpEndPoint}"),
+ new GrpcChannelOptions
+ {
+ HttpHandler = new SocketsHttpHandler
+ {
+ SslOptions = { RemoteCertificateValidationCallback = delegate { return true; } }
+ }
+ });
+ _client = new StreamsClient(_channel);
return leader;
}
@@ -235,7 +261,7 @@ private async Task RestoreNode(int nodeIndex, int coldChunkNumber)
private EndPoint[] GossipSeedsFor(int nodeIndex) =>
_nodeEndpoints
.Where((_, index) => index != nodeIndex)
- .Select(x => (EndPoint)x.HttpEndPoint)
+ .Select(x => (EndPoint)x.NodeEndPoint)
.ToArray();
private async Task WaitForArchiveCheckpoint(long minimum)
@@ -258,6 +284,7 @@ private async Task WaitForArchiveCheckpoint(long minimum)
[OneTimeTearDown]
public override async Task TestFixtureTearDown()
{
+ _channel?.Dispose();
await base.TestFixtureTearDown();
if (_s3Client is null)
{
diff --git a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs
index 6b8aa33884..b88b48a586 100644
--- a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs
+++ b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs
@@ -4,7 +4,6 @@
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
-using EventStore.ClientAPI;
using EventStore.Core.Data;
using EventStore.Core.Tests.Helpers;
using EventStore.Plugins.Subsystems;
@@ -17,7 +16,6 @@ public abstract class specification_with_cluster : Specif
{
protected MiniClusterNode[] _nodes;
protected Endpoints[] _nodeEndpoints;
- protected IEventStoreConnection _conn;
protected virtual TimeSpan GivenTimeout { get; } = TimeSpan.FromMinutes(2);
protected virtual int NodeCount => 3;
@@ -25,15 +23,13 @@ public abstract class specification_with_cluster : Specif
protected class Endpoints
{
- public readonly IPEndPoint InternalTcp;
- public readonly IPEndPoint ExternalTcp;
- public readonly IPEndPoint HttpEndPoint;
+ public readonly IPEndPoint NodeEndPoint;
+ public readonly IPEndPoint ReplicationEndPoint;
public IEnumerable Ports()
{
- yield return InternalTcp.Port;
- yield return ExternalTcp.Port;
- yield return HttpEndPoint.Port;
+ yield return NodeEndPoint.Port;
+ yield return ReplicationEndPoint.Port;
}
private readonly List _sockets;
@@ -44,21 +40,15 @@ public Endpoints()
var defaultLoopBack = new IPEndPoint(IPAddress.Loopback, 0);
- var internalTcp = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
- internalTcp.Bind(defaultLoopBack);
- _sockets.Add(internalTcp);
+ var nodeEndpoint = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+ nodeEndpoint.Bind(defaultLoopBack);
+ _sockets.Add(nodeEndpoint);
+ var replicationEndpoint = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+ replicationEndpoint.Bind(defaultLoopBack);
+ _sockets.Add(replicationEndpoint);
- var externalTcp = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
- externalTcp.Bind(defaultLoopBack);
- _sockets.Add(externalTcp);
-
- var httpEndPoint = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
- httpEndPoint.Bind(defaultLoopBack);
- _sockets.Add(httpEndPoint);
-
- InternalTcp = CopyEndpoint((IPEndPoint)internalTcp.LocalEndPoint);
- ExternalTcp = CopyEndpoint((IPEndPoint)externalTcp.LocalEndPoint);
- HttpEndPoint = CopyEndpoint((IPEndPoint)httpEndPoint.LocalEndPoint);
+ NodeEndPoint = CopyEndpoint((IPEndPoint)nodeEndpoint.LocalEndPoint);
+ ReplicationEndPoint = CopyEndpoint((IPEndPoint)replicationEndpoint.LocalEndPoint);
}
public void DisposeSockets()
@@ -102,7 +92,7 @@ public override async Task TestFixtureSetUp()
nodeIndex,
_nodeEndpoints[nodeIndex],
_nodeEndpoints.Where((_, otherIndex) => otherIndex != nodeIndex)
- .Select(x => (EndPoint)x.HttpEndPoint)
+ .Select(x => (EndPoint)x.NodeEndPoint)
.ToArray(),
wait));
_nodes[nodeIndex] = _nodeCreationFactory[nodeIndex](true);
@@ -144,9 +134,6 @@ public override async Task TestFixtureSetUp()
onFail: MiniNodeLogging.WriteLogs,
msg: $"Waiting for followers timed out! States={string.Join(", ", _nodes.Select(n => n.NodeState))}");
- _conn = CreateConnection();
- await _conn.ConnectAsync();
-
try
{
await Given().WithTimeout(GivenTimeout);
@@ -158,9 +145,6 @@ public override async Task TestFixtureSetUp()
}
}
- protected virtual IEventStoreConnection CreateConnection() =>
- EventStoreConnection.Create(_nodes[0].ExternalTcpEndPoint);
-
protected virtual void BeforeNodesStart()
{
}
@@ -171,8 +155,7 @@ protected virtual void BeforeNodesStart()
protected virtual MiniClusterNode CreateNode(int index, Endpoints endpoints, EndPoint[] gossipSeeds,
bool wait = true) => new(
- PathName, index, endpoints.InternalTcp,
- endpoints.ExternalTcp, endpoints.HttpEndPoint,
+ PathName, index, endpoints.NodeEndPoint, endpoints.ReplicationEndPoint,
subsystems: Array.Empty(), gossipSeeds: gossipSeeds);
[TearDown]
@@ -187,7 +170,6 @@ public void AfterEachTest()
[OneTimeTearDown]
public override async Task TestFixtureTearDown()
{
- _conn?.Close();
if (_nodes is not null)
{
await Task.WhenAll(_nodes.Where(node => node is not null).Select(node => node.Shutdown()));
diff --git a/src/EventStore.Core.Tests/Integration/when_cluster_nodes_are_restarted.cs b/src/EventStore.Core.Tests/Integration/when_cluster_nodes_are_restarted.cs
index 53791c1be1..ee99b6ff2d 100644
--- a/src/EventStore.Core.Tests/Integration/when_cluster_nodes_are_restarted.cs
+++ b/src/EventStore.Core.Tests/Integration/when_cluster_nodes_are_restarted.cs
@@ -102,7 +102,7 @@ private int SelectRestartNode(bool[] restartedNodes, bool restartLeader)
private EndPoint[] GossipSeedsFor(int restartedNodeIndex) =>
_nodeEndpoints
.Where((_, index) => index != restartedNodeIndex)
- .Select(x => (EndPoint)x.HttpEndPoint)
+ .Select(x => (EndPoint)x.NodeEndPoint)
.ToArray();
[Test]
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/ClusterSettingsFactory.cs b/src/EventStore.Core.Tests/Services/ElectionsService/ClusterSettingsFactory.cs
index f63c8c9319..b072443420 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/ClusterSettingsFactory.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/ClusterSettingsFactory.cs
@@ -1,9 +1,7 @@
using System;
using System.Linq;
using System.Net;
-using System.Runtime.InteropServices;
using EventStore.Core.Cluster.Settings;
-using EventStore.Core.Tests.Services.Transport.Tcp;
namespace EventStore.Core.Tests.Services.ElectionsService;
@@ -14,13 +12,9 @@ public class ClusterSettingsFactory
private static ClusterVNodeSettings CreateVNode(int nodeNumber, bool isReadOnlyReplica)
{
- int tcpIntPort = StartingPort + nodeNumber * 2,
- tcpExtPort = tcpIntPort + 1,
- httpPort = tcpIntPort + 11;
+ var httpPort = StartingPort + nodeNumber;
return new ClusterVNodeSettings(Guid.NewGuid(), 0,
- GetLoopbackForPort(tcpIntPort), null,
- GetLoopbackForPort(tcpExtPort), null,
GetLoopbackForPort(httpPort), 0,
isReadOnlyReplica);
}
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/ClusterVNodeSettings.cs b/src/EventStore.Core.Tests/Services/ElectionsService/ClusterVNodeSettings.cs
index e1fb14badc..b9c4a888d9 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/ClusterVNodeSettings.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/ClusterVNodeSettings.cs
@@ -16,24 +16,14 @@ public class ClusterVNodeSettings
public readonly bool ReadOnlyReplica;
public ClusterVNodeSettings(Guid instanceId, int debugIndex,
- IPEndPoint internalTcpEndPoint,
- IPEndPoint internalSecureTcpEndPoint,
- IPEndPoint externalTcpEndPoint,
- IPEndPoint externalSecureTcpEndPoint,
IPEndPoint httpEndPoint,
int nodePriority,
bool readOnlyReplica)
{
Ensure.NotEmptyGuid(instanceId, "instanceId");
- Ensure.Equal(false, internalTcpEndPoint == null && internalSecureTcpEndPoint == null, "Both internal TCP endpoints are null");
-
Ensure.NotNull(httpEndPoint, nameof(httpEndPoint));
- NodeInfo = new VNodeInfo(instanceId, debugIndex,
- internalTcpEndPoint, internalSecureTcpEndPoint,
- externalTcpEndPoint, externalSecureTcpEndPoint,
- httpEndPoint,
- readOnlyReplica);
+ NodeInfo = new VNodeInfo(instanceId, debugIndex, httpEndPoint, readOnlyReplica);
NodePriority = nodePriority;
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/ElectionServiceUnit.cs b/src/EventStore.Core.Tests/Services/ElectionsService/ElectionServiceUnit.cs
index b24a1e875e..a07c837474 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/ElectionServiceUnit.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/ElectionServiceUnit.cs
@@ -47,11 +47,7 @@ public ElectionsServiceUnit(ClusterSettings clusterSettings)
_bus = new(GetType().Name);
var memberInfo = MemberInfo.Initial(clusterSettings.Self.NodeInfo.InstanceId, InitialDate,
VNodeState.Unknown, true,
- clusterSettings.Self.NodeInfo.InternalTcp,
- clusterSettings.Self.NodeInfo.InternalSecureTcp,
- clusterSettings.Self.NodeInfo.ExternalTcp,
- clusterSettings.Self.NodeInfo.ExternalSecureTcp,
- clusterSettings.Self.NodeInfo.HttpEndPoint, null, 0, 0,
+ clusterSettings.Self.NodeInfo.HttpEndPoint, null, 0,
clusterSettings.Self.NodePriority,
clusterSettings.Self.ReadOnlyReplica);
ElectionsService = new Core.Services.ElectionsService(Publisher,
@@ -83,11 +79,7 @@ private ClusterInfo BuildClusterInfo(ClusterSettings clusterSettings)
InitialDate,
VNodeState.Unknown,
true,
- clusterSettings.Self.NodeInfo.InternalTcp,
- clusterSettings.Self.NodeInfo.InternalSecureTcp,
- clusterSettings.Self.NodeInfo.ExternalTcp,
- clusterSettings.Self.NodeInfo.ExternalSecureTcp,
- clusterSettings.Self.NodeInfo.HttpEndPoint, null, 0, 0,
+ clusterSettings.Self.NodeInfo.HttpEndPoint, null, 0,
LastCommitPosition, WriterCheckpoint, ChaserCheckpoint,
-1,
-1,
@@ -98,11 +90,7 @@ private ClusterInfo BuildClusterInfo(ClusterSettings clusterSettings)
InitialDate,
VNodeState.Unknown,
true,
- x.NodeInfo.InternalTcp,
- x.NodeInfo.InternalSecureTcp,
- x.NodeInfo.ExternalTcp,
- x.NodeInfo.ExternalSecureTcp,
- x.NodeInfo.HttpEndPoint, null, 0, 0,
+ x.NodeInfo.HttpEndPoint, null, 0,
LastCommitPosition, WriterCheckpoint, ChaserCheckpoint,
-1,
-1,
@@ -196,9 +184,7 @@ public IEnumerable ListMembers(Func predicate = nu
? MemberInfo.ForManager(x.InstanceId, x.TimeStamp, x.IsAlive,
x.HttpEndPoint)
: MemberInfo.ForVNode(x.InstanceId, x.TimeStamp, x.State, x.IsAlive,
- x.InternalTcpEndPoint, x.InternalSecureTcpEndPoint,
- x.ExternalTcpEndPoint, x.ExternalSecureTcpEndPoint,
- x.HttpEndPoint, null, 0, 0,
+ x.HttpEndPoint, null, 0,
x.LastCommitPosition, x.WriterCheckpoint, x.ChaserCheckpoint,
x.EpochPosition, x.EpochNumber, x.EpochId, x.NodePriority, x.IsReadOnlyReplica));
}
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/ElectionsServiceTests.cs b/src/EventStore.Core.Tests/Services/ElectionsService/ElectionsServiceTests.cs
index bcf9e0938f..378cc8075c 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/ElectionsServiceTests.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/ElectionsServiceTests.cs
@@ -29,18 +29,12 @@ public abstract class ElectionsFixture
protected static Func NodeFactory = (id) => new VNodeInfo(
Guid.Parse($"00000000-0000-0000-0000-00000000000{id}"), id,
- new IPEndPoint(IPAddress.Loopback, id),
- new IPEndPoint(IPAddress.Loopback, id),
- new IPEndPoint(IPAddress.Loopback, id),
- new IPEndPoint(IPAddress.Loopback, id),
new IPEndPoint(IPAddress.Loopback, id), false);
protected static readonly Func MemberInfoFromVNode =
(nodeInfo, timestamp, state, isAlive, epochNumber, epochId, priority) => MemberInfo.ForVNode(
nodeInfo.InstanceId, timestamp, state, isAlive,
- nodeInfo.InternalTcp,
- nodeInfo.InternalSecureTcp, nodeInfo.ExternalTcp, nodeInfo.ExternalSecureTcp,
- nodeInfo.HttpEndPoint, null, 0, 0,
+ nodeInfo.HttpEndPoint, null, 0,
0, 0, 0, 0, epochNumber, epochId, priority,
nodeInfo.IsReadOnlyReplica);
@@ -924,9 +918,7 @@ public void should_send_an_acceptance_to_other_members()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeThree.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeThree.InternalTcp,
- _nodeThree.InternalSecureTcp, _nodeThree.ExternalTcp, _nodeThree.ExternalSecureTcp,
- _nodeThree.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, 0, _epochId, 0,
+ _nodeThree.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, _epochId, 0,
_nodeThree.IsReadOnlyReplica)),
new GrpcMessage.SendOverGrpc(_nodeThree.HttpEndPoint,
new ElectionMessage.Accept(_node.InstanceId, _node.HttpEndPoint,
@@ -1077,9 +1069,7 @@ public void should_complete_elections()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, 0, _epochId, 0,
+ _nodeTwo.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
_publisher.Messages.Should().BeEquivalentTo(expected);
@@ -1314,9 +1304,7 @@ public void should_attempt_not_to_elect_previously_elected_leader()
new ElectionMessage.ElectionsDone(3,1,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, 0, _epochId, 0,
+ _nodeTwo.HttpEndPoint, null, 0, 0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
_publisher.Messages.Should().BeEquivalentTo(expected);
@@ -1357,11 +1345,7 @@ public void should_throw_argument_exception()
var endpoint = new IPEndPoint(IPAddress.Loopback, 1234);
var nodeInfo = MemberInfo.Initial(Guid.NewGuid(),
DateTime.UtcNow, VNodeState.ReadOnlyLeaderless, true,
- endpoint,
- endpoint,
- endpoint,
- endpoint,
- endpoint, null, 0, 0,
+ endpoint, null, 0,
0,
true);
@@ -1405,9 +1389,7 @@ public void previous_leader_should_be_elected()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeThree.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeThree.InternalTcp,
- _nodeThree.InternalSecureTcp, _nodeThree.ExternalTcp, _nodeThree.ExternalSecureTcp,
- _nodeThree.HttpEndPoint, null, 0, 0,
+ _nodeThree.HttpEndPoint, null, 0,
0, 0, 0, 0, 0, _epochId, 0,
_nodeThree.IsReadOnlyReplica)),
};
@@ -1505,9 +1487,7 @@ public void previous_leader_should_not_be_elected()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0,
+ _nodeTwo.HttpEndPoint, null, 0,
0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
@@ -1541,9 +1521,7 @@ public void previous_leader_should_not_be_elected()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0,
+ _nodeTwo.HttpEndPoint, null, 0,
0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
@@ -1598,9 +1576,7 @@ public void previous_leader_should_not_be_elected()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0,
+ _nodeTwo.HttpEndPoint, null, 0,
0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
@@ -1634,9 +1610,7 @@ public void previous_leader_should_not_be_elected()
new ElectionMessage.ElectionsDone(0,0,
MemberInfo.ForVNode(
_nodeTwo.InstanceId, _timeProvider.UtcNow, VNodeState.Unknown, true,
- _nodeTwo.InternalTcp,
- _nodeTwo.InternalSecureTcp, _nodeTwo.ExternalTcp, _nodeTwo.ExternalSecureTcp,
- _nodeTwo.HttpEndPoint, null, 0, 0,
+ _nodeTwo.HttpEndPoint, null, 0,
0, 0, 0, 0, 0, _epochId, 0,
_nodeTwo.IsReadOnlyReplica)),
};
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/LeaderNode/ElectionsServiceUnitTests.cs b/src/EventStore.Core.Tests/Services/ElectionsService/LeaderNode/ElectionsServiceUnitTests.cs
index 11c8462e3d..6e3a013cf5 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/LeaderNode/ElectionsServiceUnitTests.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/LeaderNode/ElectionsServiceUnitTests.cs
@@ -40,7 +40,7 @@ public void Setup()
seeds.Add(endPoint);
var instanceId = Guid.Parse($"101EFD13-F9CD-49BE-9C6D-E6AF9AF5540{i}");
var memberInfo = MemberInfo.ForVNode(instanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- endPoint, null, endPoint, null, endPoint, null, 0, 0, -1, 0, 0, -1, -1, Guid.Empty, 0, false);
+ endPoint, null, 0, -1, 0, 0, -1, -1, Guid.Empty, 0, false);
members.Add(memberInfo);
_fakeTimeProvider = new FakeTimeProvider();
_scheduler = new FakeScheduler(new FakeTimer(), _fakeTimeProvider);
@@ -379,7 +379,7 @@ Func epochNumber
{
var id = IdForNode(i);
var ep = EndpointForNode(i);
- return MemberInfo.ForVNode(id, DateTime.Now, VNodeState.Follower, true, ep, ep, ep, ep, ep, null, 0, 0,
+ return MemberInfo.ForVNode(id, DateTime.Now, VNodeState.Follower, true, ep, null, 0,
-1, writerCheckpoint(i), chaserCheckpoint(i), 1, epochNumber(i), epochId, nodePriority(i), false);
}
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/RandomizedElectionsTestCase.cs b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/RandomizedElectionsTestCase.cs
index 68b6ce5fd9..168f65a466 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/RandomizedElectionsTestCase.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/RandomizedElectionsTestCase.cs
@@ -74,7 +74,7 @@ public virtual void Init()
var outputBus = new SynchronousScheduler($"ELECTIONS-OUTPUT-BUS-{i}");
var endPoint = new IPEndPoint(BaseEndPoint.Address, BaseEndPoint.Port + i);
var memberInfo = MemberInfo.Initial(Guid.NewGuid(), DateTime.UtcNow, VNodeState.Unknown, true,
- endPoint, endPoint, endPoint, endPoint, endPoint, null, 0, 0, 0, false);
+ endPoint, null, 0, 0, false);
_instances.Add(new ElectionsInstance(memberInfo.InstanceId, endPoint, inputBus, outputBus));
sendOverHttpHandler.RegisterEndPoint(endPoint, inputBus);
@@ -125,8 +125,7 @@ protected virtual GossipMessage.GossipUpdated GetInitialGossipFor(ElectionsInsta
{
var members = allInstances.Select(
x => MemberInfo.ForVNode(x.InstanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- x.EndPoint, null, x.EndPoint, null,
- x.EndPoint, null, 0, 0, -1, 0, 0, -1, -1, Guid.Empty, 0, false));
+ x.EndPoint, null, 0, -1, 0, 0, -1, -1, Guid.Empty, 0, false));
var gossip = new GossipMessage.GossipUpdated(new ClusterInfo(members.ToArray()));
return gossip;
}
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/UpdateGossipProcessor.cs b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/UpdateGossipProcessor.cs
index 7ed5cd5fc9..b9381ca8dc 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/UpdateGossipProcessor.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/UpdateGossipProcessor.cs
@@ -63,7 +63,7 @@ public void Process(int iteration, RandTestQueueItem item)
previousMembers[leaderIndex] =
MemberInfo.ForVNode(previousLeaderInfo.InstanceId, DateTime.UtcNow, VNodeState.Leader,
- previousLeaderInfo.IsAlive, leaderEndPoint, null, leaderEndPoint, null, leaderEndPoint, null, 0, 0,
+ previousLeaderInfo.IsAlive, leaderEndPoint, null, 0,
-1, 0, 0, -1, -1, Guid.Empty, 0, false);
}
}
@@ -82,7 +82,7 @@ public void Process(int iteration, RandTestQueueItem item)
foreach (var memberInfo in updatedGossip)
{
- _sendOverGrpcProcessor.RegisterEndpointToSkip(memberInfo.ExternalTcpEndPoint, !memberInfo.IsAlive);
+ _sendOverGrpcProcessor.RegisterEndpointToSkip(memberInfo.HttpEndPoint, !memberInfo.IsAlive);
}
var updateGossipMessage = new GossipMessage.GossipUpdated(new ClusterInfo(updatedGossip));
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_full_imediately.cs b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_full_imediately.cs
index 339ca873bd..6722010f0f 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_full_imediately.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_full_imediately.cs
@@ -37,7 +37,7 @@ private MemberInfo[] CreateInitialGossip(ElectionsInstance instance, ElectionsIn
{
return new[] {
MemberInfo.ForVNode(instance.InstanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- instance.EndPoint, null, instance.EndPoint, null, instance.EndPoint, null, 0, 0,
+ instance.EndPoint, null, 0,
-1, 0, 0, -1, -1, Guid.Empty, 0, false)
};
}
@@ -53,7 +53,7 @@ private MemberInfo[] CreateUpdatedGossip(int iteration,
Console.WriteLine("Update item: {0} : {1}", iteration, item.EndPoint.GetPort());
return instances.Select((x, i) =>
MemberInfo.ForVNode(x.InstanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- x.EndPoint, null, x.EndPoint, null, x.EndPoint, null, 0, 0,
+ x.EndPoint, null, 0,
-1, 0, 0, -1, -1, Guid.Empty, 0, false)).ToArray();
}
diff --git a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_to_full_later.cs b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_to_full_later.cs
index 05f165cfab..cdf621e6ab 100644
--- a/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_to_full_later.cs
+++ b/src/EventStore.Core.Tests/Services/ElectionsService/Randomized/elections_service_5_nodes_with_1_known_when_started_and_set_to_full_later.cs
@@ -37,7 +37,7 @@ private MemberInfo[] CreateInitialGossip(ElectionsInstance instance, ElectionsIn
{
return new[] {
MemberInfo.ForVNode(instance.InstanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- instance.EndPoint, null, instance.EndPoint, null, instance.EndPoint, null, 0, 0,
+ instance.EndPoint, null, 0,
-1, 0, 0, -1, -1, Guid.Empty, 0, false)
};
}
@@ -57,7 +57,7 @@ private MemberInfo[] CreateUpdatedGossip(int iteration,
{
return instances.Select((x, i) =>
MemberInfo.ForVNode(x.InstanceId, DateTime.UtcNow, VNodeState.Unknown, true,
- x.EndPoint, null, x.EndPoint, null, x.EndPoint, null, 0, 0,
+ x.EndPoint, null, 0,
-1, 0, 0, -1, -1, Guid.Empty, 0, false))
.ToArray();
}
diff --git a/src/EventStore.Core.Tests/Services/GossipService/NodeGossipServiceTests.cs b/src/EventStore.Core.Tests/Services/GossipService/NodeGossipServiceTests.cs
index 11ebbfbf88..2c877c6645 100644
--- a/src/EventStore.Core.Tests/Services/GossipService/NodeGossipServiceTests.cs
+++ b/src/EventStore.Core.Tests/Services/GossipService/NodeGossipServiceTests.cs
@@ -43,31 +43,15 @@ public NodeGossipServiceTestFixture()
_currentNode = new VNodeInfo(
Guid.Parse("00000000-0000-0000-0000-000000000001"), 1,
- new IPEndPoint(IPAddress.Loopback, 1111),
- new IPEndPoint(IPAddress.Loopback, 1111),
- new IPEndPoint(IPAddress.Loopback, 1111),
- new IPEndPoint(IPAddress.Loopback, 1111),
new IPEndPoint(IPAddress.Loopback, 1111), false);
_nodeTwo = new VNodeInfo(
Guid.Parse("00000000-0000-0000-0000-000000000002"), 2,
- new IPEndPoint(IPAddress.Loopback, 2222),
- new IPEndPoint(IPAddress.Loopback, 2222),
- new IPEndPoint(IPAddress.Loopback, 2222),
- new IPEndPoint(IPAddress.Loopback, 2222),
new IPEndPoint(IPAddress.Loopback, 2222), false);
_nodeThree = new VNodeInfo(
Guid.Parse("00000000-0000-0000-0000-000000000003"), 3,
- new IPEndPoint(IPAddress.Loopback, 3333),
- new IPEndPoint(IPAddress.Loopback, 3333),
- new IPEndPoint(IPAddress.Loopback, 3333),
- new IPEndPoint(IPAddress.Loopback, 3333),
new IPEndPoint(IPAddress.Loopback, 3333), false);
_nodeFour = new VNodeInfo(
Guid.Parse("00000000-0000-0000-0000-000000000004"), 4,
- new IPEndPoint(IPAddress.Loopback, 4444),
- new IPEndPoint(IPAddress.Loopback, 4444),
- new IPEndPoint(IPAddress.Loopback, 4444),
- new IPEndPoint(IPAddress.Loopback, 4444),
new IPEndPoint(IPAddress.Loopback, 4444), false);
_getNodeToGossipTo = infos => infos.First(x => Equals(x.HttpEndPoint, _nodeTwo.HttpEndPoint));
@@ -120,8 +104,7 @@ protected static MemberInfo MemberInfoForVNode(VNodeInfo nodeInfo, DateTime utcN
VNodeState nodeState = VNodeState.Initializing, string esVersion = VersionInfo.DefaultVersion, bool isAlive = true)
{
return MemberInfo.ForVNode(nodeInfo.InstanceId, utcNow, nodeState, isAlive,
- nodeInfo.InternalTcp, nodeInfo.InternalSecureTcp, nodeInfo.ExternalTcp,
- nodeInfo.ExternalSecureTcp, nodeInfo.HttpEndPoint, null, 0, 0,
+ nodeInfo.HttpEndPoint, null, 0,
0, writerCheckpoint ?? 0, 0, -1, epochNumber ?? -1, Guid.Empty, nodePriority ?? 0, false, esVersion);
}
@@ -743,7 +726,7 @@ protected override Message[] Given() =>
_nodeTwo.HttpEndPoint);
[Test]
- public void should_ignore_message_and_wait_for_tcp_to_decide()
+ public void should_ignore_message_and_wait_for_connection_state_to_change()
{
ExpectNoMessages();
}
@@ -1031,7 +1014,7 @@ private static MemberInfo TestNodeFor(int identifier, bool isAlive, DateTime tim
{
var ipEndpoint = new IPEndPoint(IPAddress.Loopback, identifier);
return MemberInfo.ForVNode(Guid.NewGuid(), timeStamp, VNodeState.Initializing, isAlive,
- ipEndpoint, ipEndpoint, ipEndpoint, ipEndpoint, ipEndpoint, null, 0, 0,
+ ipEndpoint, null, 0,
0, 0, 0, -1, -1, Guid.Empty, 0, false);
}
@@ -1108,7 +1091,7 @@ private static MemberInfo TestNodeFor(int identifier, bool isAlive, DateTime tim
{
var ipEndpoint = new IPEndPoint(IPAddress.Loopback, identifier);
return MemberInfo.ForVNode(Guid.NewGuid(), timeStamp, nodeState, isAlive,
- ipEndpoint, ipEndpoint, ipEndpoint, ipEndpoint, ipEndpoint, null, 0, 0,
+ ipEndpoint, null, 0,
0, 0, 0, -1, -1, Guid.Empty, 0, false);
}
diff --git a/src/EventStore.Core.Tests/Services/Replication/LogReplication/LogReplicationFixture.cs b/src/EventStore.Core.Tests/Services/Replication/LogReplication/LogReplicationFixture.cs
index a7514a3a51..1dfac124f9 100644
--- a/src/EventStore.Core.Tests/Services/Replication/LogReplication/LogReplicationFixture.cs
+++ b/src/EventStore.Core.Tests/Services/Replication/LogReplication/LogReplicationFixture.cs
@@ -157,14 +157,9 @@ private async ValueTask> CreateLeader(TFChunkDb db, Cancel
timeStamp: DateTime.Now,
state: VNodeState.Leader,
isAlive: true,
- internalTcpEndPoint: FakeEndPoint,
- internalSecureTcpEndPoint: null,
- externalTcpEndPoint: null,
- externalSecureTcpEndPoint: null,
httpEndPoint: FakeEndPoint,
advertiseHostToClientAs: null,
advertiseHttpPortToClientAs: 0,
- advertiseTcpPortToClientAs: 0,
lastCommitPosition: 0,
writerCheckpoint: 0,
chaserCheckpoint: 0,
diff --git a/src/EventStore.Core.Tests/Services/Replication/ReadOnlyReplica/connecting_to_read_only_replica.cs b/src/EventStore.Core.Tests/Services/Replication/ReadOnlyReplica/connecting_to_read_only_replica.cs
index f8e76a11ab..9faa415dc9 100644
--- a/src/EventStore.Core.Tests/Services/Replication/ReadOnlyReplica/connecting_to_read_only_replica.cs
+++ b/src/EventStore.Core.Tests/Services/Replication/ReadOnlyReplica/connecting_to_read_only_replica.cs
@@ -1,11 +1,18 @@
+using System;
using System.Net;
+using System.Net.Http;
+using System.Text;
using System.Threading.Tasks;
-using EventStore.ClientAPI;
-using EventStore.ClientAPI.Exceptions;
-using EventStore.Core.Tests.ClientAPI.Helpers;
+using EventStore.Client.Streams;
+using EventStore.Core.Services.Transport.Grpc;
using EventStore.Core.Tests.Helpers;
using EventStore.Core.Tests.Integration;
+using Google.Protobuf;
+using Grpc.Core;
+using Grpc.Net.Client;
using NUnit.Framework;
+using Empty = EventStore.Client.Empty;
+using GrpcMetadata = EventStore.Core.Services.Transport.Grpc.Constants.Metadata;
namespace EventStore.Core.Tests.Replication.ReadOnlyReplica;
@@ -13,13 +20,15 @@ namespace EventStore.Core.Tests.Replication.ReadOnlyReplica;
[TestFixture(typeof(LogFormat.V2), typeof(string))]
public class connecting_to_read_only_replica : specification_with_cluster
{
+ protected override async Task Given() =>
+ await _nodes[2].AdminUserCreated.WithTimeout(TimeSpan.FromSeconds(30));
+
protected override MiniClusterNode CreateNode(int index, Endpoints endpoints, EndPoint[] gossipSeeds,
bool wait = true)
{
var isReadOnly = index == 2;
var node = new MiniClusterNode(
- PathName, index, endpoints.InternalTcp,
- endpoints.ExternalTcp, endpoints.HttpEndPoint, gossipSeeds,
+ PathName, index, endpoints.NodeEndPoint, endpoints.ReplicationEndPoint, gossipSeeds,
readOnlyReplica: isReadOnly);
if (wait && !isReadOnly)
{
@@ -29,35 +38,82 @@ protected override MiniClusterNode CreateNode(int index,
return node;
}
- protected override IEventStoreConnection CreateConnection()
+ private static CallOptions GetCallOptions()
{
- var settings = ConnectionSettings.Create()
- .DisableServerCertificateValidation()
- .PerformOnAnyNode();
- return EventStoreConnection.Create(settings, _nodes[2].ExternalTcpEndPoint);
+ var credentials = CallCredentials.FromInterceptor((_, metadata) =>
+ {
+ metadata.Add("authorization",
+ $"Basic {Convert.ToBase64String(Encoding.ASCII.GetBytes("admin:changeit"))}");
+ return Task.CompletedTask;
+ });
+ return new CallOptions(credentials: credentials, deadline: DateTime.UtcNow.AddSeconds(30));
}
- [Test]
- public async Task append_to_stream_should_fail_with_not_supported_exception()
+ private Streams.StreamsClient CreateClient(out GrpcChannel channel, out HttpClient httpClient)
{
- const string stream = "append_to_stream_should_fail_with_not_supported_exception";
- await AssertEx.ThrowsAsync(
- () => _conn.AppendToStreamAsync(stream, ExpectedVersion.Any, TestEvent.NewTestEvent()));
+ httpClient = new HttpClient(new SocketsHttpHandler
+ {
+ SslOptions = { RemoteCertificateValidationCallback = delegate { return true; } }
+ });
+ channel = GrpcChannel.ForAddress(new Uri($"https://{_nodes[2].HttpEndPoint}"),
+ new GrpcChannelOptions { HttpClient = httpClient });
+ return new Streams.StreamsClient(channel);
}
[Test]
- public async Task delete_stream_should_fail_with_not_supported_exception()
+ public async Task append_to_stream_is_rejected()
{
- const string stream = "delete_stream_should_fail_with_not_supported_exception";
- await AssertEx.ThrowsAsync(() =>
- _conn.DeleteStreamAsync(stream, ExpectedVersion.Any));
+ var client = CreateClient(out var channel, out var httpClient);
+ using (channel)
+ using (httpClient)
+ using (var call = client.Append(GetCallOptions()))
+ {
+ await call.RequestStream.WriteAsync(new AppendReq
+ {
+ Options = new()
+ {
+ Any = new Empty(),
+ StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(nameof(append_to_stream_is_rejected)) }
+ }
+ });
+ await call.RequestStream.WriteAsync(new AppendReq
+ {
+ ProposedMessage = new()
+ {
+ Id = Uuid.NewUuid().ToDto(),
+ Data = ByteString.Empty,
+ CustomMetadata = ByteString.Empty,
+ Metadata =
+ {
+ [GrpcMetadata.Type] = "test",
+ [GrpcMetadata.ContentType] = GrpcMetadata.ContentTypes.ApplicationJson
+ }
+ }
+ });
+ await call.RequestStream.CompleteAsync();
+
+ var exception = Assert.ThrowsAsync(async () => await call.ResponseAsync);
+ Assert.That(exception.StatusCode, Is.EqualTo(StatusCode.NotFound));
+ }
}
[Test]
- public async Task start_transaction_should_fail_with_not_supported_exception()
+ public async Task delete_stream_is_rejected()
{
- const string stream = "start_transaction_should_fail_with_not_supported_exception";
- await AssertEx.ThrowsAsync(() =>
- _conn.StartTransactionAsync(stream, ExpectedVersion.Any));
+ var client = CreateClient(out var channel, out var httpClient);
+ using (channel)
+ using (httpClient)
+ using (var call = client.DeleteAsync(new DeleteReq
+ {
+ Options = new()
+ {
+ Any = new Empty(),
+ StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(nameof(delete_stream_is_rejected)) }
+ }
+ }, GetCallOptions()))
+ {
+ var exception = Assert.ThrowsAsync(async () => await call.ResponseAsync);
+ Assert.That(exception.StatusCode, Is.EqualTo(StatusCode.NotFound));
+ }
}
}
diff --git a/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingServiceTests.cs b/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingServiceTests.cs
index e861b74159..70d7ef4a93 100644
--- a/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingServiceTests.cs
+++ b/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingServiceTests.cs
@@ -791,10 +791,6 @@ private static MemberInfo CreateLeader(Guid? instanceId = null, int httpPort = 2
DateTime.UtcNow,
VNodeState.Leader,
true,
- new DnsEndPoint("leader-replication.internal", 1112),
- new DnsEndPoint("leader-replication.internal", 1113),
- null,
- null,
new DnsEndPoint("leader.internal", httpPort),
null,
0,
@@ -803,7 +799,6 @@ private static MemberInfo CreateLeader(Guid? instanceId = null, int httpPort = 2
0,
0,
0,
- 0,
Guid.NewGuid(),
0,
false);
diff --git a/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingTransportSecurityTests.cs b/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingTransportSecurityTests.cs
index 17fe60ab3f..50186afc83 100644
--- a/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingTransportSecurityTests.cs
+++ b/src/EventStore.Core.Tests/Services/RequestForwarding/GrpcRequestForwardingTransportSecurityTests.cs
@@ -53,7 +53,7 @@ public async Task cleartext_credential_rejection_completes_the_proxy_correlation
supervisor.Handle(new ClientMessage.ForwardMessage(request));
- var response = publisher.Messages.OfType().Single();
+ var response = publisher.Messages.OfType().Single();
Assert.That(response.CorrelationId, Is.EqualTo(request.InternalCorrId));
}
@@ -119,10 +119,6 @@ private static MemberInfo CreateLeader() => MemberInfo.ForVNode(
DateTime.UtcNow,
VNodeState.Leader,
true,
- new DnsEndPoint("leader-replication.internal", 1112),
- new DnsEndPoint("leader-replication.internal", 1113),
- null,
- null,
new DnsEndPoint("leader.internal", 2113),
null,
0,
@@ -131,7 +127,6 @@ private static MemberInfo CreateLeader() => MemberInfo.ForVNode(
0,
0,
0,
- 0,
Guid.NewGuid(),
0,
false);
diff --git a/src/EventStore.Core.Tests/Services/RequestForwarding/RequestForwardingServiceTests.cs b/src/EventStore.Core.Tests/Services/RequestForwarding/RequestForwardingServiceTests.cs
index d0ed4eb5e5..8eb4dccb3c 100644
--- a/src/EventStore.Core.Tests/Services/RequestForwarding/RequestForwardingServiceTests.cs
+++ b/src/EventStore.Core.Tests/Services/RequestForwarding/RequestForwardingServiceTests.cs
@@ -55,13 +55,13 @@ public void not_authenticated_survives_the_client_correlation_rewrite()
clientCorrelationId,
new CallbackEnvelope(message => response = message),
TimeSpan.FromMinutes(1),
- new TcpMessage.NotAuthenticated(clientCorrelationId, "timeout"));
+ new ClientMessage.NotAuthenticated(clientCorrelationId, "timeout"));
var service = new RequestForwardingService(
new NoopPublisher(), forwardingProxy, TimeSpan.FromSeconds(1));
- service.Handle(new TcpMessage.NotAuthenticated(internalCorrelationId, "not authenticated"));
+ service.Handle(new ClientMessage.NotAuthenticated(internalCorrelationId, "not authenticated"));
- var completion = (TcpMessage.NotAuthenticated)response;
+ var completion = (ClientMessage.NotAuthenticated)response;
Assert.Multiple(() =>
{
Assert.That(completion.CorrelationId, Is.EqualTo(clientCorrelationId));
diff --git a/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader.cs b/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader.cs
index 8ef3d8d03b..6f3bf4ff30 100644
--- a/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader.cs
+++ b/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader.cs
@@ -42,10 +42,6 @@ private static MemberInfo FakeMemberInfo()
return EventStore.Core.Cluster.MemberInfo.Initial(Guid.Empty, DateTime.UtcNow,
VNodeState.Unknown, true,
new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- null, 0, 0, 0, false);
+ null, 0, 0, false);
}
}
diff --git a/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader_and_replica_moves_forward.cs b/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader_and_replica_moves_forward.cs
index 994a62f28d..4b7ba26312 100644
--- a/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader_and_replica_moves_forward.cs
+++ b/src/EventStore.Core.Tests/Services/RequestManagement/Service/when_writing_and_deposed_as_leader_and_replica_moves_forward.cs
@@ -37,10 +37,6 @@ private static MemberInfo FakeMemberInfo()
return EventStore.Core.Cluster.MemberInfo.Initial(Guid.Empty, DateTime.UtcNow,
VNodeState.Unknown, true,
new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- new IPEndPoint(IPAddress.Parse(ipAddress), port),
- null, 0, 0, 0, false);
+ null, 0, 0, false);
}
}
diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/Forwarding/ForwardingGrpcCodecTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/Forwarding/ForwardingGrpcCodecTests.cs
index 828810fab5..e4396f5693 100644
--- a/src/EventStore.Core.Tests/Services/Transport/Grpc/Forwarding/ForwardingGrpcCodecTests.cs
+++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/Forwarding/ForwardingGrpcCodecTests.cs
@@ -415,8 +415,6 @@ public void not_handled_leader_info_round_trips()
Guid.NewGuid(),
ClientMessage.NotHandled.Types.NotHandledReason.NotLeader,
new ClientMessage.NotHandled.Types.LeaderInfo(
- new DnsEndPoint("leader-tcp.internal", 1113),
- true,
new DnsEndPoint("leader-http.internal", 2113)));
var decoded = RoundTripResponse(message);
@@ -425,28 +423,6 @@ public void not_handled_leader_info_round_trips()
{
Assert.That(decoded.CorrelationId, Is.EqualTo(message.CorrelationId));
Assert.That(decoded.Reason, Is.EqualTo(message.Reason));
- Assert.That(decoded.LeaderInfo.IsSecure, Is.True);
- Assert.That(decoded.LeaderInfo.ExternalTcp, Is.EqualTo(message.LeaderInfo.ExternalTcp));
- Assert.That(decoded.LeaderInfo.Http, Is.EqualTo(message.LeaderInfo.Http));
- });
- }
-
- [Test]
- public void not_handled_leader_info_without_external_tcp_round_trips_as_null()
- {
- var message = new ClientMessage.NotHandled(
- Guid.NewGuid(),
- ClientMessage.NotHandled.Types.NotHandledReason.NotLeader,
- new ClientMessage.NotHandled.Types.LeaderInfo(
- null,
- false,
- new DnsEndPoint("leader-http.internal", 2113)));
-
- var decoded = RoundTripResponse(message);
-
- Assert.Multiple(() =>
- {
- Assert.That(decoded.LeaderInfo.ExternalTcp, Is.Null);
Assert.That(decoded.LeaderInfo.Http, Is.EqualTo(message.LeaderInfo.Http));
});
}
@@ -454,9 +430,9 @@ public void not_handled_leader_info_without_external_tcp_round_trips_as_null()
[Test]
public void not_authenticated_round_trips()
{
- var message = new TcpMessage.NotAuthenticated(Guid.NewGuid(), "not authenticated");
+ var message = new ClientMessage.NotAuthenticated(Guid.NewGuid(), "not authenticated");
- var decoded = RoundTripResponse(message);
+ var decoded = RoundTripResponse(message);
Assert.Multiple(() =>
{
diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceFactoryTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceFactoryTests.cs
index 19697515ac..91ba8c4a35 100644
--- a/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceFactoryTests.cs
+++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceFactoryTests.cs
@@ -37,7 +37,30 @@ public void replication_client_uses_node_certificate_names_and_owns_its_http_cli
Assert.That(nodeHttpClientFactory.AdditionalCertificateNames, Is.EqualTo(new[] { "cluster.internal" }));
client.Dispose();
- Assert.That(nodeHttpClientFactory.Handler.Disposed, Is.True);
+ Assert.That(nodeHttpClientFactory.TransportHandler.Disposed, Is.True);
+ }
+
+ [Test]
+ public void replication_client_normalizes_legacy_sub_second_heartbeat_values()
+ {
+ var nodeHttpClientFactory = new RecordingNodeHttpClientFactory();
+ var factory = new ReplicationGrpcClientFactory(
+ Uri.UriSchemeHttps,
+ nodeHttpClientFactory,
+ TimeSpan.FromMilliseconds(700),
+ TimeSpan.FromMilliseconds(700));
+
+ using var client = factory.Create(new DnsEndPoint("leader.internal", 1112));
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(nodeHttpClientFactory.SocketsHandler.KeepAlivePingDelay,
+ Is.EqualTo(TimeSpan.FromSeconds(1)));
+ Assert.That(nodeHttpClientFactory.SocketsHandler.KeepAlivePingTimeout,
+ Is.EqualTo(TimeSpan.FromSeconds(1)));
+ Assert.That(nodeHttpClientFactory.SocketsHandler.KeepAlivePingPolicy,
+ Is.EqualTo(HttpKeepAlivePingPolicy.Always));
+ });
}
[Test]
@@ -99,7 +122,8 @@ private sealed class TrackingReplicationGrpcClientFactory : IReplicationGrpcClie
private sealed class RecordingNodeHttpClientFactory : INodeHttpClientFactory
{
- public RecordingHttpMessageHandler Handler { get; } = new();
+ public RecordingHttpMessageHandler TransportHandler { get; } = new();
+ public SocketsHttpHandler SocketsHandler { get; } = new();
public string[] AdditionalCertificateNames { get; private set; }
public HttpClient CreateHttpClient(
@@ -107,7 +131,8 @@ public HttpClient CreateHttpClient(
Action configureSocketsHttpHandler = null)
{
AdditionalCertificateNames = additionalCertificateNames;
- return new HttpClient(Handler);
+ configureSocketsHttpHandler?.Invoke(SocketsHandler);
+ return new HttpClient(TransportHandler);
}
}
diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceSupervisorTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceSupervisorTests.cs
index a6eaedf281..dcd662b4f8 100644
--- a/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceSupervisorTests.cs
+++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/Replication/GrpcReplicaServiceSupervisorTests.cs
@@ -22,7 +22,7 @@ public class GrpcReplicaServiceSupervisorTests
[TestCase(false)]
[TestCase(true)]
- public async Task pre_replica_state_starts_and_tracks_a_stream_for_the_advertised_http_endpoints(
+ public async Task pre_replica_state_starts_and_tracks_a_stream_for_the_advertised_replication_endpoints(
bool readOnlyReplica)
{
var fixture = CreateFixture();
@@ -37,7 +37,7 @@ public async Task pre_replica_state_starts_and_tracks_a_stream_for_the_advertise
var request = fixture.Factory.Requests.Single();
Assert.Multiple(() =>
{
- Assert.That(request.Endpoints.LeaderEndPoint, Is.EqualTo(fixture.Leader.HttpEndPoint));
+ Assert.That(request.Endpoints.LeaderEndPoint, Is.EqualTo(fixture.Leader.ReplicationEndPoint));
Assert.That(request.Endpoints.AdvertisedReplicaEndPoint, Is.EqualTo(fixture.AdvertisedEndPoint));
Assert.That(request.Service.StartCalls, Is.EqualTo(1));
Assert.That(fixture.TrackedTasks.Single(), Is.SameAs(request.Service.Task));
@@ -410,7 +410,7 @@ private static Fixture CreateFixture(
startException,
createException,
beforeCreateReturns);
- var advertisedEndPoint = new DnsEndPoint("replica.internal", 2113);
+ var advertisedEndPoint = new DnsEndPoint("replica.internal", 1112);
var trackedTasks = new List();
var supervisor = new GrpcReplicaServiceSupervisor(
publisher,
@@ -432,10 +432,6 @@ private static MemberInfo CreateLeader() => MemberInfo.ForVNode(
DateTime.UtcNow,
VNodeState.Leader,
true,
- new DnsEndPoint("leader-replication.internal", 1112),
- null,
- null,
- null,
new DnsEndPoint("leader.internal", 2113),
null,
0,
@@ -444,10 +440,10 @@ private static MemberInfo CreateLeader() => MemberInfo.ForVNode(
0,
0,
0,
- 0,
Guid.NewGuid(),
0,
- false);
+ false,
+ replicationEndPoint: new DnsEndPoint("leader.replication.internal", 1112));
private static SystemMessage.StateChangeMessage CreateReplicaState(
VNodeState state,
diff --git a/src/EventStore.Core.Tests/Services/Transport/Tcp/core_tcp_package.cs b/src/EventStore.Core.Tests/Services/Transport/Tcp/core_tcp_package.cs
deleted file mode 100644
index ffcb22e037..0000000000
--- a/src/EventStore.Core.Tests/Services/Transport/Tcp/core_tcp_package.cs
+++ /dev/null
@@ -1,199 +0,0 @@
-using System;
-using EventStore.Core.Services.Transport.Tcp;
-using NUnit.Framework;
-
-namespace EventStore.Core.Tests.Services.Transport.Tcp;
-
-[TestFixture]
-public class core_tcp_package
-{
- [Test]
- public void should_throw_argument_null_exception_when_created_as_authorized_but_login_not_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, Guid.NewGuid(), null, "pa$$",
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_null_exception_when_created_as_authorized_but_password_not_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, Guid.NewGuid(), "login", null,
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_null_exception_when_created_as_authorized_but_token_not_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, Guid.NewGuid(), null,
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_exception_when_created_as_not_authorized_but_login_is_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.None, Guid.NewGuid(), "login", null,
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_exception_when_created_as_not_authorized_but_password_is_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.None, Guid.NewGuid(), null, "pa$$",
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_exception_when_created_as_not_authorized_but_token_is_provided()
- {
- Assert.Throws(() =>
- new TcpPackage(TcpCommand.BadRequest, TcpFlags.None, Guid.NewGuid(), "token",
- new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void not_authorized_with_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.None, corrId, null, null, new byte[] { 1, 2, 3 });
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.None, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.False(pkg.Tokens.TryGetValue("uid", out _));
- Assert.False(pkg.Tokens.TryGetValue("pwd", out _));
- Assert.False(pkg.Tokens.TryGetValue("jwt", out _));
-
- Assert.AreEqual(3, pkg.Data.Count);
- Assert.AreEqual(1, pkg.Data.Array[pkg.Data.Offset + 0]);
- Assert.AreEqual(2, pkg.Data.Array[pkg.Data.Offset + 1]);
- Assert.AreEqual(3, pkg.Data.Array[pkg.Data.Offset + 2]);
- }
-
- [Test]
- public void not_authorized_with_empty_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.None, corrId, null, null, new byte[0]);
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.None, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.False(pkg.Tokens.TryGetValue("uid", out _));
- Assert.False(pkg.Tokens.TryGetValue("pwd", out _));
- Assert.False(pkg.Tokens.TryGetValue("jwt", out _));
-
- Assert.AreEqual(0, pkg.Data.Count);
- }
-
- [Test]
- public void authorized_with_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, corrId, "login", "pa$$",
- new byte[] { 1, 2, 3 });
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.Authenticated, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.AreEqual("login", pkg.Tokens["uid"]);
- Assert.AreEqual("pa$$", pkg.Tokens["pwd"]);
- Assert.False(pkg.Tokens.TryGetValue("jwt", out _));
-
- Assert.AreEqual(3, pkg.Data.Count);
- Assert.AreEqual(1, pkg.Data.Array[pkg.Data.Offset + 0]);
- Assert.AreEqual(2, pkg.Data.Array[pkg.Data.Offset + 1]);
- Assert.AreEqual(3, pkg.Data.Array[pkg.Data.Offset + 2]);
- }
-
- [Test]
- public void authorized_with_empty_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, corrId, "login", "pa$$",
- new byte[0]);
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.Authenticated, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.AreEqual("login", pkg.Tokens["uid"]);
- Assert.AreEqual("pa$$", pkg.Tokens["pwd"]);
- Assert.False(pkg.Tokens.TryGetValue("jwt", out _));
-
- Assert.AreEqual(0, pkg.Data.Count);
- }
-
- [Test]
- public void token_authorized_with_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, corrId, "token",
- new byte[] { 1, 2, 3 });
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.Authenticated, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.AreEqual("token", pkg.Tokens["jwt"]);
- Assert.False(pkg.Tokens.TryGetValue("uid", out _));
- Assert.False(pkg.Tokens.TryGetValue("pwd", out _));
-
- Assert.AreEqual(3, pkg.Data.Count);
- Assert.AreEqual(1, pkg.Data.Array[pkg.Data.Offset + 0]);
- Assert.AreEqual(2, pkg.Data.Array[pkg.Data.Offset + 1]);
- Assert.AreEqual(3, pkg.Data.Array[pkg.Data.Offset + 2]);
- }
-
- [Test]
- public void token_authorized_with_empty_data_should_serialize_and_deserialize_correctly()
- {
- var corrId = Guid.NewGuid();
- var refPkg = new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated, corrId, "token",
- new byte[0]);
- var bytes = refPkg.AsArraySegment();
-
- var pkg = TcpPackage.FromArraySegment(bytes);
- Assert.AreEqual(TcpCommand.BadRequest, pkg.Command);
- Assert.AreEqual(TcpFlags.Authenticated, pkg.Flags);
- Assert.AreEqual(corrId, pkg.CorrelationId);
- Assert.AreEqual("token", pkg.Tokens["jwt"]);
- Assert.False(pkg.Tokens.TryGetValue("uid", out _));
- Assert.False(pkg.Tokens.TryGetValue("pwd", out _));
-
- Assert.AreEqual(0, pkg.Data.Count);
- }
-
- [Test]
- public void should_throw_argument_exception_when_login_too_long()
- {
- Assert.Throws(() => new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated,
- Guid.NewGuid(), new string('*', TcpPackage.MaxLoginLength + 1), "pa$$", new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_exception_when_password_too_long()
- {
- Assert.Throws(() => new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated,
- Guid.NewGuid(), "login", new string('*', TcpPackage.MaxPasswordLength + 1), new byte[] { 1, 2, 3 }));
- }
-
- [Test]
- public void should_throw_argument_exception_when_token_too_long()
- {
- Assert.Throws(() => new TcpPackage(TcpCommand.BadRequest, TcpFlags.Authenticated,
- Guid.NewGuid(), new string('*', TcpPackage.MaxTokenLength + 1), new byte[] { 1, 2, 3 }));
- }
-}
diff --git a/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connection.cs b/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connection.cs
deleted file mode 100644
index f6165d0f82..0000000000
--- a/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connection.cs
+++ /dev/null
@@ -1,174 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Net;
-using System.Reflection;
-using System.Security.Cryptography;
-using System.Security.Cryptography.X509Certificates;
-using System.Text;
-using System.Threading;
-using EventStore.Common.Utils;
-using EventStore.Core.Services.Transport.Tcp;
-using EventStore.Core.Tests.Helpers;
-using EventStore.Transport.Tcp;
-using NUnit.Framework;
-using ILogger = Serilog.ILogger;
-
-namespace EventStore.Core.Tests.Services.Transport.Tcp;
-
-[TestFixture]
-public class ssl_connections
-{
- private static readonly ILogger Log = Serilog.Log.ForContext();
- private IPAddress _ip;
- private int _port;
-
- [SetUp]
- public void SetUp()
- {
- _ip = IPAddress.Loopback;
- _port = PortsHelper.GetAvailablePort(_ip);
- }
-
- [Test]
- public void should_connect_to_each_other_and_send_data()
- {
- var serverEndPoint = new IPEndPoint(_ip, _port);
- X509Certificate2 cert = GetServerCertificate();
-
- var sent = new byte[1000];
- new Random().NextBytes(sent);
-
- using var received = new MemoryStream();
-
- var done = new ManualResetEventSlim();
-
- var listener = new TcpServerListener(serverEndPoint);
- listener.StartListening((endPoint, socket) =>
- {
- var ssl = TcpConnectionSsl.CreateServerFromSocket(Guid.NewGuid(), endPoint, socket,
- () => cert, null, delegate
- { return (true, null); },
- verbose: true);
- ssl.ConnectionClosed += (x, y) => done.Set();
- if (ssl.IsClosed)
- {
- done.Set();
- }
-
- Action>> callback = null;
- callback = (x, y) =>
- {
- foreach (var arraySegment in y)
- {
- received.Write(arraySegment.Array, arraySegment.Offset, arraySegment.Count);
- Log.Information("Received: {0} bytes, total: {1}.", arraySegment.Count, received.Length);
- }
-
- if (received.Length >= sent.Length)
- {
- Log.Information("Done receiving...");
- done.Set();
- }
- else
- {
- Log.Information("Receiving...");
- ssl.ReceiveAsync(callback);
- }
- };
- Log.Information("Receiving...");
- ssl.ReceiveAsync(callback);
- }, "Secure");
-
- var clientSsl = TcpConnectionSsl.CreateConnectingConnection(
- Guid.NewGuid(),
- serverEndPoint.GetHost(),
- null,
- serverEndPoint,
- delegate
- { return (true, null); },
- null,
- new TcpClientConnector(),
- TcpConnectionManager.ConnectionTimeout,
- conn =>
- {
- Log.Information("Sending bytes...");
- conn.EnqueueSend(new[] { new ArraySegment(sent) });
- },
- (conn, err) =>
- {
- Log.Error("Connecting failed: {0}.", err);
- done.Set();
- },
- verbose: true);
-
- Assert.IsTrue(done.Wait(20000), "Took too long to receive completion.");
-
- Log.Information("Stopping listener...");
- listener.Stop();
- Log.Information("Closing client TLS connection...");
- clientSsl.Close("Normal close.");
- Log.Information("Checking received data...");
- Assert.AreEqual(sent, received.ToArray());
- }
-
- private static X509Certificate2 _root, _server, _otherServer, _untrusted;
- public static X509Certificate2 GetRootCertificate()
- {
- _root ??= GetCertificate("ca", loadKey: false);
- return new X509Certificate2(_root);
- }
-
- public static X509Certificate2 GetServerCertificate()
- {
- _server ??= GetCertificate("node1");
- return new X509Certificate2(_server);
- }
-
- public static X509Certificate2 GetOtherServerCertificate()
- {
- _otherServer ??= GetCertificate("node2");
- return new X509Certificate2(_otherServer);
- }
-
- public static X509Certificate2 GetUntrustedCertificate()
- {
- _untrusted ??= GetCertificate("untrusted");
- return new X509Certificate2(_untrusted);
- }
-
- private static X509Certificate2 GetCertificate(string name, bool loadKey = true)
- {
- const string resourcePath = "EventStore.Core.Tests.Services.Transport.Tcp.test_certificates";
-
- var certBytes = LoadResource($"{resourcePath}.{name}.{name}.crt");
- var certificate = X509Certificate2.CreateFromPem(Encoding.UTF8.GetString(certBytes));
-
- if (!loadKey)
- {
- return certificate;
- }
-
- var keyBytes = LoadResource($"{resourcePath}.{name}.{name}.key");
- using var rsa = RSA.Create();
- rsa.ImportFromPem(Encoding.UTF8.GetString(keyBytes));
-
- using X509Certificate2 certWithKey = certificate.CopyWithPrivateKey(rsa);
-
- // recreate the certificate from a PKCS #12 bundle to work around: https://github.com/dotnet/runtime/issues/23749
- return X509CertificateLoader.LoadPkcs12(certWithKey.ExportToPkcs12(), string.Empty, X509KeyStorageFlags.Exportable);
- }
-
- private static byte[] LoadResource(string resource)
- {
- using var resourceStream = Assembly.GetExecutingAssembly().GetManifestResourceStream(resource);
- if (resourceStream == null)
- {
- return null;
- }
-
- using var memStream = new MemoryStream();
- resourceStream.CopyTo(memStream);
- return memStream.ToArray();
- }
-}
diff --git a/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connections_mutual_auth.cs b/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connections_mutual_auth.cs
deleted file mode 100644
index 1a80ae435e..0000000000
--- a/src/EventStore.Core.Tests/Services/Transport/Tcp/ssl_connections_mutual_auth.cs
+++ /dev/null
@@ -1,149 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Net;
-using System.Net.Security;
-using System.Reflection;
-using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using EventStore.Common.Utils;
-using EventStore.Core.Services.Transport.Tcp;
-using EventStore.Core.Tests.Helpers;
-using EventStore.Transport.Tcp;
-using NUnit.Framework;
-using Serilog;
-
-namespace EventStore.Core.Tests.Services.Transport.Tcp;
-
-[TestFixture]
-public class ssl_connections_mutual_auth
-{
- private static readonly ILogger Log = Serilog.Log.ForContext();
- private IPAddress _ip;
- private int _port;
-
- [SetUp]
- public void SetUp()
- {
- _ip = IPAddress.Loopback;
- _port = PortsHelper.GetAvailablePort(_ip);
- }
-
- [TestCase(true, true, true, true, true)] //require valid server and client certificate
- [TestCase(true, false, true, true, false)] //require valid server and client certificate
- [TestCase(false, true, true, true, false)] //require valid server and client certificate
- [TestCase(false, false, true, true, false)] //require valid server and client certificate
- [TestCase(true, true, true, false, true)] //require valid server certificate only
- [TestCase(true, false, true, false, true)] //require valid server certificate only
- [TestCase(false, true, true, false, false)] //require valid server certificate only
- [TestCase(false, false, true, false, false)] //require valid server certificate only
- [TestCase(true, true, false, true, true)] //require valid client certificate only
- [TestCase(true, false, false, true, false)] //require valid client certificate only
- [TestCase(false, true, false, true, true)] //require valid client certificate only
- [TestCase(false, false, false, true, false)] //require valid client certificate only
- [TestCase(true, true, false, false, true)] //do not require valid client or server certificate
- [TestCase(true, false, false, false, true)] //do not require valid client or server certificate
- [TestCase(false, true, false, false, true)] //do not require valid client or server certificate
- [TestCase(false, false, false, false, true)] //do not require valid client or server certificate
- public void should_connect_to_each_other_and_send_data_depending_on_certificate_validity_and_settings(
- bool useValidServerCertificate,
- bool useValidClientCertificate,
- bool validateServerCertificate,
- bool validateClientCertificate,
- bool shouldConnectSuccessfully
- )
- {
- var serverEndPoint = new IPEndPoint(_ip, _port);
- var serverCertificate = useValidServerCertificate
- ? ssl_connections.GetServerCertificate()
- : ssl_connections.GetUntrustedCertificate();
- var clientCertificate = useValidClientCertificate
- ? ssl_connections.GetOtherServerCertificate()
- : ssl_connections.GetUntrustedCertificate();
- var rootCertificates = new X509Certificate2Collection(ssl_connections.GetRootCertificate());
-
-
- var sent = new byte[1000];
- new Random().NextBytes(sent);
-
- var received = new MemoryStream();
-
- var done = new ManualResetEventSlim();
-
- var listener = new TcpServerListener(serverEndPoint);
- listener.StartListening((endPoint, socket) =>
- {
- var ssl = TcpConnectionSsl.CreateServerFromSocket(Guid.NewGuid(), endPoint, socket, () => serverCertificate,
- null, (cert, chain, err) => validateClientCertificate ?
- ClusterVNode.ValidateClientCertificate(cert, chain, err, () => null, () => rootCertificates) : (true, null),
- verbose: true);
- ssl.ConnectionClosed += (x, y) => done.Set();
- if (ssl.IsClosed)
- {
- done.Set();
- }
-
- Action>> callback = null;
- callback = (x, y) =>
- {
- foreach (var arraySegment in y)
- {
- received.Write(arraySegment.Array, arraySegment.Offset, arraySegment.Count);
- Log.Information("Received: {0} bytes, total: {1}.", arraySegment.Count, received.Length);
- }
-
- if (received.Length >= sent.Length)
- {
- Log.Information("Done receiving...");
- done.Set();
- }
- else
- {
- Log.Information("Receiving...");
- ssl.ReceiveAsync(callback);
- }
- };
- Log.Information("Receiving...");
- ssl.ReceiveAsync(callback);
- }, "Secure");
-
- var clientSsl = TcpConnectionSsl.CreateConnectingConnection(
- Guid.NewGuid(),
- serverEndPoint.GetHost(),
- null,
- serverEndPoint,
- (cert, chain, err, _) => validateServerCertificate ? ClusterVNode.ValidateServerCertificate(cert, chain, err, () => null, () => rootCertificates, null) : (true, null),
- () => new X509CertificateCollection { clientCertificate },
- new TcpClientConnector(),
- TcpConnectionManager.ConnectionTimeout,
- conn =>
- {
- Log.Information("Sending bytes...");
- conn.EnqueueSend(new[] { new ArraySegment(sent) });
- },
- (conn, err) =>
- {
- Log.Error("Connecting failed: {0}.", err);
- done.Set();
- },
- verbose: true);
-
- Assert.IsTrue(done.Wait(20000), "Took too long to receive completion.");
-
- Log.Information("Stopping listener...");
- listener.Stop();
- Log.Information("Closing client TLS connection...");
- clientSsl.Close("Normal close.");
- Log.Information("Checking received data...");
-
- if (shouldConnectSuccessfully)
- {
- Assert.AreEqual(sent, received.ToArray());
- }
- else
- {
- Assert.AreEqual(new byte[0], received.ToArray());
- }
- }
-}
diff --git a/src/EventStore.Core.Tests/Services/VNode/InaugurationManager/InaugurationManagerTests.cs b/src/EventStore.Core.Tests/Services/VNode/InaugurationManager/InaugurationManagerTests.cs
index 8ae7429f0c..fc9ce572a8 100644
--- a/src/EventStore.Core.Tests/Services/VNode/InaugurationManager/InaugurationManagerTests.cs
+++ b/src/EventStore.Core.Tests/Services/VNode/InaugurationManager/InaugurationManagerTests.cs
@@ -20,8 +20,7 @@ public abstract class InaugurationManagerTests
protected readonly MemberInfo _leader =
MemberInfo.ForVNode(
default, default, default, default,
- new DnsEndPoint("localhost", default), default, default, default,
- new DnsEndPoint("localhost", default), default, default, default,
+ new DnsEndPoint("localhost", default), default, default,
default, default, default, default, default, default, default, default);
protected readonly long _replicationTarget = 400;
protected readonly long _indexTarget = 400;
diff --git a/src/EventStore.Core.Tests/Services/VNode/ShutdownServiceTests.cs b/src/EventStore.Core.Tests/Services/VNode/ShutdownServiceTests.cs
index f2adb06c98..d1011fa934 100644
--- a/src/EventStore.Core.Tests/Services/VNode/ShutdownServiceTests.cs
+++ b/src/EventStore.Core.Tests/Services/VNode/ShutdownServiceTests.cs
@@ -1,5 +1,4 @@
using System;
-using System.Net;
using DotNext.Net.Http;
using EventStore.Core.Data;
using EventStore.Core.Messages;
@@ -16,10 +15,6 @@ public class ShutdownServiceTests
= new(
Guid.NewGuid(),
0,
- new IPEndPoint(0, 0),
- new IPEndPoint(IPAddress.Loopback, 1),
- new IPEndPoint(IPAddress.Loopback, 2),
- new IPEndPoint(IPAddress.Loopback, 3),
new HttpEndPoint(new Uri("http://www.trogondb.com")), true);
[Test]
diff --git a/src/EventStore.Core.Tests/Services/VNode/leader_info_provider.cs b/src/EventStore.Core.Tests/Services/VNode/leader_info_provider.cs
index 68de1f0112..9a34d143df 100644
--- a/src/EventStore.Core.Tests/Services/VNode/leader_info_provider.cs
+++ b/src/EventStore.Core.Tests/Services/VNode/leader_info_provider.cs
@@ -1,8 +1,6 @@
using System;
using System.Collections.Generic;
-using System.Linq;
using System.Net;
-using EventStore.Common.Utils;
using EventStore.Core.Cluster;
using EventStore.Core.Data;
using EventStore.Core.Services.VNode;
@@ -13,252 +11,45 @@ namespace EventStore.Core.Tests.Services.VNode;
[TestFixture]
public class leader_info_provider
{
- private const string DefaultHttpEndPoint = "9.9.9.9:9";
-
- public static IEnumerable |