From 19e88e3b054b949bb235bfc609245c83353abd26 Mon Sep 17 00:00:00 2001 From: Ente_3 Date: Wed, 30 Sep 2026 17:36:25 +0200 Subject: [PATCH 1/2] fix: reconnect cluster nodes after a lost connection A dropped node connection removed both nodes from the cluster right away, and nothing tried to reconnect. A lost connection now enters the reconnect grace period if cloudnet.max.node.disconnect.millis is greater than 0. The default (0) keeps the old behaviour. - only one reconnect attempt per node at a time - remove nodes stuck in syncing once the grace period expires - send the startup time on authorization, so a restart during the grace period is treated as a new node - remote service providers look up the current node channel on each call instead of keeping the one from registration - apply the reconnecting node's services during service sync - flush queued packets and re-publish local services after a plain authorization - re-select the head node when a node becomes ready again - read the grace period when NodeDisconnectHandler is created, not at class load, so tests can set it - don't track reconnect attempts for the local node - add Javadoc (@since 4.0) to the new public methods in NodeDisconnectHandler --- .../java22/src/main/resources/launcher.cnl | 8 +- .../task/NodeDisconnectTrackerTask.java | 73 +++++++++++++++---- .../cluster/util/NodeDisconnectHandler.java | 54 ++++++++++++++ .../DefaultNetworkClientChannelHandler.java | 13 +++- .../DefaultNetworkServerChannelHandler.java | 6 +- .../listener/AuthorizationPacketListener.java | 17 +++++ .../AuthorizationResponsePacketListener.java | 30 +++++++- .../defaults/DefaultCloudServiceManager.java | 15 +++- .../src/main/resources/lang/de_DE.properties | 1 + .../src/main/resources/lang/en_US.properties | 1 + 10 files changed, 189 insertions(+), 29 deletions(-) diff --git a/launcher/java22/src/main/resources/launcher.cnl b/launcher/java22/src/main/resources/launcher.cnl index 2d77ecd804..1b2922ee5d 100644 --- a/launcher/java22/src/main/resources/launcher.cnl +++ b/launcher/java22/src/main/resources/launcher.cnl @@ -85,9 +85,11 @@ var cloudnet.node.memory 256 # setting for this) #var cloudnet.max.node.idle.millis 30000 # Sets the timeout until a node gets force disconnected from the current node after getting marked as disconnected (see -# the previous setting). This defaults to 0 as we want a node to get disconnected instantly. Increasing this timeout might -# lead to unexpected cluster behaviour, such as registered players which are already disconnected or registered services -# which were already stopped during the disconnect time. USE THIS SETTING AT YOUR OWN RISK! +# the previous setting) or after the connection to the node was lost. During this time the nodes try to reconnect to each +# other. This defaults to 0 as we want a node to get disconnected instantly. Increasing this timeout (for example to 30000 +# for nodes in different datacenters) might lead to unexpected cluster behaviour, such as registered players which are +# already disconnected or registered services which were already stopped during the disconnect time. +# USE THIS SETTING AT YOUR OWN RISK! #var cloudnet.max.node.disconnect.millis 0 # Sets the path to the directory in which the local xodus database file is located #var cloudnet.database.xodus.path local/database/xodus diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTask.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTask.java index 4cd499e50b..1ceb1e996e 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTask.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTask.java @@ -18,30 +18,52 @@ import eu.cloudnetservice.driver.language.I18n; import eu.cloudnetservice.driver.registry.Service; +import eu.cloudnetservice.node.cluster.NodeServer; import eu.cloudnetservice.node.cluster.NodeServerProvider; import eu.cloudnetservice.node.cluster.NodeServerState; -import eu.cloudnetservice.node.impl.cluster.util.QueuedNetworkChannel; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; import eu.cloudnetservice.utils.base.concurrent.TaskUtil; +import jakarta.inject.Inject; import jakarta.inject.Singleton; import java.time.Duration; import java.time.Instant; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import lombok.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @Singleton -public record NodeDisconnectTrackerTask(@NonNull @Service I18n i18n, @NonNull NodeServerProvider provider) implements Runnable { +public final class NodeDisconnectTrackerTask implements Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(NodeDisconnectTrackerTask.class); private static final long SOFT_DISCONNECT_MS_DELAY = Long.getLong("cloudnet.max.node.idle.millis", 30_000); - private static final long HARD_DISCONNECT_MS_DELAY = Long.getLong("cloudnet.max.node.disconnect.millis", 0); + // the time to wait for an established reconnect connection to complete the handshake before trying again + private static final Duration RECONNECT_HANDSHAKE_TIMEOUT = Duration.ofSeconds(15); + + private final I18n i18n; + private final NodeServerProvider provider; + private final NodeDisconnectHandler disconnectHandler; + private final Map pendingReconnects = new ConcurrentHashMap<>(); + + @Inject + public NodeDisconnectTrackerTask( + @NonNull @Service I18n i18n, + @NonNull NodeServerProvider provider, + @NonNull NodeDisconnectHandler disconnectHandler + ) { + this.i18n = i18n; + this.provider = provider; + this.disconnectHandler = disconnectHandler; + } @Override public void run() { try { var currentTime = Instant.now(); var local = this.provider.localNode(); + var hardDisconnectMillis = this.disconnectHandler.hardDisconnectMillis(); // first check all currently connected nodes if they are idling for too long for (var server : this.provider.nodeServers()) { // ignore the local node and all nodes which are not yet ready (these nodes do nothing which can lead to errors in @@ -54,12 +76,7 @@ public void run() { var updateDelay = Duration.between(server.lastNodeInfoUpdate(), currentTime).toMillis(); if (updateDelay >= SOFT_DISCONNECT_MS_DELAY) { // the node is idling for too long! Mark the node as disconnected and begin to schedule all packets to the node - server.state(NodeServerState.DISCONNECTED); - server.channel(new QueuedNetworkChannel(server.channel())); - // trigger a head node refresh if the server is the head node to ensure that we're not using a head node which is dead - if (this.provider.headNode().equals(server)) { - this.provider.selectHeadNode(); - } + this.disconnectHandler.markNodeServerDisconnected(server); // warn about that LOGGER.warn(this.i18n.translate("cluster-server-soft-disconnect", server.name(), updateDelay)); } @@ -67,26 +84,32 @@ public void run() { // now check if a node is idling for ages and hard disconnect them for (var server : this.provider.nodeServers()) { - // skip the local node and all nodes which aren't mark as disconnected (yet) - if (server == local || server.state() != NodeServerState.DISCONNECTED) { + // skip the local node and all nodes which aren't mark as disconnected (yet). nodes which are syncing are still in + // the reconnect process and must be disconnected as well if the reconnect doesn't complete in time + if (server == local) { + continue; + } + var state = server.state(); + if (state != NodeServerState.DISCONNECTED && state != NodeServerState.SYNCING) { + this.pendingReconnects.remove(server.name()); continue; } // check if the node is exceeding the hard disconnect delay var disconnectMs = Duration.between(server.lastStateChange(), currentTime).toMillis(); - if (disconnectMs >= HARD_DISCONNECT_MS_DELAY) { + if (disconnectMs >= hardDisconnectMillis) { // close hard + this.pendingReconnects.remove(server.name()); server.close(); LOGGER.warn(this.i18n.translate( "cluster-server-hard-disconnect", server.name(), - HARD_DISCONNECT_MS_DELAY, + hardDisconnectMillis, disconnectMs)); - } else { + } else if (state == NodeServerState.DISCONNECTED) { // check if we need to reconnect or if the other node is responsible to reconnect if (local.nodeInfoSnapshot().startupMillis() > server.nodeInfoSnapshot().startupMillis()) { - // try to connect to the node server - TaskUtil.getOrDefault(server.connect(), Duration.ofSeconds(5), null); + this.tryReconnect(server, currentTime); } } } @@ -94,4 +117,22 @@ public void run() { LOGGER.error("Exception ticking node disconnect tracker", exception); } } + + private void tryReconnect(@NonNull NodeServer server, @NonNull Instant currentTime) { + // the reconnect handshake can't handle multiple concurrent connections to the same node, don't open + // another connection while the last established one might still be waiting for its handshake. a + // state change after the last connection means that it either completed or failed + var lastReconnect = this.pendingReconnects.get(server.name()); + if (lastReconnect != null + && lastReconnect.isAfter(server.lastStateChange()) + && Duration.between(lastReconnect, currentTime).compareTo(RECONNECT_HANDSHAKE_TIMEOUT) < 0) { + return; + } + + // try to connect to the node server + var connected = TaskUtil.getOrDefault(server.connect().thenApply(_ -> true), Duration.ofSeconds(5), false); + if (connected) { + this.pendingReconnects.put(server.name(), Instant.now()); + } + } } diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandler.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandler.java index e797d598a4..20dd253198 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandler.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandler.java @@ -26,6 +26,7 @@ import eu.cloudnetservice.driver.service.ServiceInfoSnapshot; import eu.cloudnetservice.driver.service.ServiceLifeCycle; import eu.cloudnetservice.node.cluster.NodeServer; +import eu.cloudnetservice.node.cluster.NodeServerState; import eu.cloudnetservice.node.impl.service.InternalCloudServiceManager; import eu.cloudnetservice.node.service.CloudService; import jakarta.inject.Inject; @@ -43,6 +44,7 @@ public final class NodeDisconnectHandler { private final I18n i18n; private final EventManager eventManager; private final InternalCloudServiceManager serviceManager; + private final long hardDisconnectMillis; @Inject public NodeDisconnectHandler( @@ -53,6 +55,7 @@ public NodeDisconnectHandler( this.i18n = i18n; this.eventManager = eventManager; this.serviceManager = serviceManager; + this.hardDisconnectMillis = Long.getLong("cloudnet.max.node.disconnect.millis", 0); } private static @NonNull ChannelMessage.Builder targetServices(@NonNull Collection services) { @@ -65,6 +68,57 @@ public NodeDisconnectHandler( return builder; } + /** + * Gets the time in milliseconds a node gets to reconnect after it was marked as disconnected, before it gets removed + * from the cluster. A value of 0 means that nodes are removed from the cluster instantly. The value is configured + * using the {@code cloudnet.max.node.disconnect.millis} system property. + * + * @return the reconnect grace period of nodes in milliseconds. + * @since 4.0 + */ + public long hardDisconnectMillis() { + return this.hardDisconnectMillis; + } + + /** + * Marks the given node server as disconnected. All packets sent to the node are queued until the node reconnects, + * and a new head node is selected if the given node was the head node. The node disconnect tracker takes care of + * reconnecting to the node or removing it from the cluster if it does not reconnect in time. + * + * @param server the node server to mark as disconnected. + * @throws NullPointerException if the given server is null. + * @since 4.0 + */ + public void markNodeServerDisconnected(@NonNull NodeServer server) { + // mark the node as disconnected and begin to schedule all packets to the node until it reconnected + server.state(NodeServerState.DISCONNECTED); + server.channel(new QueuedNetworkChannel(server.channel())); + // trigger a head node refresh if the server is the head node to ensure that we're not using a head node which is dead + if (server.head()) { + server.provider().selectHeadNode(); + } + } + + /** + * Handles the close of the network channel of the given node server. If a reconnect grace period is configured and + * the node is available, the node is marked as disconnected to give it the chance to reconnect. In all other cases + * the node is closed and removed from the cluster instantly. + * + * @param server the node server whose network channel was closed. + * @throws NullPointerException if the given server is null. + * @since 4.0 + */ + public void handleNodeServerChannelClose(@NonNull NodeServer server) { + // give the node the chance to reconnect if a grace period is configured, the disconnect tracker takes + // care of the reconnect and removes the node from the cluster if it did not reconnect in time + if (this.hardDisconnectMillis > 0 && server.available()) { + this.markNodeServerDisconnected(server); + LOGGER.warn(this.i18n.translate("cluster-server-connection-lost", server.name(), this.hardDisconnectMillis)); + } else { + server.close(); + } + } + public void handleNodeServerClose(@NonNull NodeServer server) { for (var snapshot : this.serviceManager.services()) { if (snapshot.serviceId().nodeUniqueId().equalsIgnoreCase(server.name())) { diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkClientChannelHandler.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkClientChannelHandler.java index f7d171871b..4ce207d69f 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkClientChannelHandler.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkClientChannelHandler.java @@ -30,6 +30,7 @@ import eu.cloudnetservice.node.cluster.NodeServerProvider; import eu.cloudnetservice.node.cluster.NodeServerState; import eu.cloudnetservice.node.config.Configuration; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; import eu.cloudnetservice.node.impl.network.listener.AuthorizationResponsePacketListener; import jakarta.inject.Inject; import jakarta.inject.Singleton; @@ -49,6 +50,7 @@ public final class DefaultNetworkClientChannelHandler implements NetworkChannelH private final NodeNetworkUtil networkUtil; private final Configuration configuration; private final NodeServerProvider nodeServerProvider; + private final NodeDisconnectHandler disconnectHandler; @Inject public DefaultNetworkClientChannelHandler( @@ -56,13 +58,15 @@ public DefaultNetworkClientChannelHandler( @NonNull EventManager eventManager, @NonNull NodeNetworkUtil networkUtil, @NonNull Configuration configuration, - @NonNull NodeServerProvider nodeServerProvider + @NonNull NodeServerProvider nodeServerProvider, + @NonNull NodeDisconnectHandler disconnectHandler ) { this.i18n = i18n; this.eventManager = eventManager; this.networkUtil = networkUtil; this.configuration = configuration; this.nodeServerProvider = nodeServerProvider; + this.disconnectHandler = disconnectHandler; } @Override @@ -72,12 +76,13 @@ public void handleChannelInitialize(@NonNull NetworkChannel channel) { channel.packetRegistry().addListener( NetworkConstants.INTERNAL_AUTHORIZATION_CHANNEL, AuthorizationResponsePacketListener.class); - // send the authentication request + // send the authentication request, the startup time allows the other node to detect restarts of this node channel.sendPacket(new AuthorizationPacket( AuthorizationPacket.PacketAuthorizationType.NODE_TO_NODE, DataBuf.empty() .writeUniqueId(this.configuration.clusterConfig().clusterId()) - .writeObject(this.configuration.identity()))); + .writeObject(this.configuration.identity()) + .writeLong(this.nodeServerProvider.localNode().nodeInfoSnapshot().startupMillis()))); LOGGER.debug(this.i18n.translate("client-network-channel-init", channel.serverAddress(), @@ -103,7 +108,7 @@ public void handleChannelClose(@NonNull NetworkChannel channel) { var clusterNodeServer = this.nodeServerProvider.node(channel); if (clusterNodeServer != null && clusterNodeServer.state() != NodeServerState.DISCONNECTED) { - clusterNodeServer.close(); + this.disconnectHandler.handleNodeServerChannelClose(clusterNodeServer); } } } diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkServerChannelHandler.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkServerChannelHandler.java index 142b33a941..161502159d 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkServerChannelHandler.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/DefaultNetworkServerChannelHandler.java @@ -29,6 +29,7 @@ import eu.cloudnetservice.node.cluster.NodeServerProvider; import eu.cloudnetservice.node.cluster.NodeServerState; import eu.cloudnetservice.node.config.Configuration; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; import eu.cloudnetservice.node.impl.network.listener.AuthorizationPacketListener; import eu.cloudnetservice.node.impl.service.InternalCloudService; import eu.cloudnetservice.node.impl.util.IpAllowlist; @@ -49,6 +50,7 @@ public final class DefaultNetworkServerChannelHandler implements NetworkChannelH private final NodeNetworkUtil networkUtil; private final Configuration configuration; private final NodeServerProvider nodeServerProvider; + private final NodeDisconnectHandler disconnectHandler; private final CloudServiceManager cloudServiceManager; private IpAllowlist ipAllowlist; @@ -60,6 +62,7 @@ public DefaultNetworkServerChannelHandler( @NonNull NodeNetworkUtil networkUtil, @NonNull Configuration configuration, @NonNull NodeServerProvider nodeServerProvider, + @NonNull NodeDisconnectHandler disconnectHandler, @NonNull CloudServiceManager cloudServiceManager ) { this.i18n = i18n; @@ -67,6 +70,7 @@ public DefaultNetworkServerChannelHandler( this.networkUtil = networkUtil; this.configuration = configuration; this.nodeServerProvider = nodeServerProvider; + this.disconnectHandler = disconnectHandler; this.cloudServiceManager = cloudServiceManager; } @@ -120,7 +124,7 @@ public void handleChannelClose(@NonNull NetworkChannel channel) { var nodeServer = this.nodeServerProvider.node(channel); if (nodeServer != null && nodeServer.state() != NodeServerState.DISCONNECTED) { - nodeServer.close(); + this.disconnectHandler.handleNodeServerChannelClose(nodeServer); } } diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListener.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListener.java index cb82c57b11..d190c3f561 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListener.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListener.java @@ -34,6 +34,7 @@ import eu.cloudnetservice.node.event.network.NetworkClusterNodeAuthSuccessEvent; import eu.cloudnetservice.node.event.network.NetworkClusterNodeReconnectEvent; import eu.cloudnetservice.node.event.network.NetworkServiceAuthSuccessEvent; +import eu.cloudnetservice.node.impl.cluster.util.QueuedNetworkChannel; import eu.cloudnetservice.node.impl.network.NodeNetworkUtil; import eu.cloudnetservice.node.impl.network.packet.AuthorizationResponsePacket; import eu.cloudnetservice.node.impl.service.InternalCloudService; @@ -88,6 +89,8 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { // read the required data for the node auth var clusterId = content.readUniqueId(); var node = content.readObject(NetworkClusterNode.class); + // the startup time of the node is not sent by nodes running an older version + var startupMillis = content.readableBytes() > 0 ? content.readLong() : -1; // check if the cluster id matches if (!this.configuration.clusterConfig().clusterId().equals(clusterId)) { break; @@ -95,6 +98,12 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { // search for the node server which represents the connected node and initialize it for (var server : this.nodeServerProvider.nodeServers()) { if (server.info().uniqueId().equals(node.uniqueId())) { + // check if the node was restarted since the last connection, in that case all data of the previous + // session is gone and the node must join the cluster like a newly started node instead of reconnecting + var lastSnapshot = server.nodeInfoSnapshot(); + if (lastSnapshot != null && startupMillis != -1 && lastSnapshot.startupMillis() != startupMillis) { + server.close(); + } // add the required packet listeners this.networkUtil.addDefaultPacketListeners(channel.packetRegistry()); channel.packetRegistry().removeListeners(NetworkConstants.INTERNAL_AUTHORIZATION_CHANNEL); @@ -113,9 +122,17 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { } else { // reply with a default auth success channel.sendPacket(new AuthorizationResponsePacket(true, false, null)); + // a previous reconnect of the node did not complete, flush the packets that were queued + // for the node and close the old channel before switching to the new channel + if (server.channel() instanceof QueuedNetworkChannel queuedChannel) { + queuedChannel.drainPacketQueue(channel); + queuedChannel.close(); + } // set the state of the node for further handling server.channel(channel); server.state(NodeServerState.READY); + // re-select the head node in case the node was the head node before it disconnected + this.nodeServerProvider.selectHeadNode(); // call the auth success event this.eventManager.callEvent(new NetworkClusterNodeAuthSuccessEvent(server, channel)); } diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListener.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListener.java index 5c4c451953..3bb2020a97 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListener.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListener.java @@ -16,6 +16,7 @@ package eu.cloudnetservice.node.impl.network.listener; +import eu.cloudnetservice.driver.channel.ChannelMessage; import eu.cloudnetservice.driver.impl.network.NetworkConstants; import eu.cloudnetservice.driver.language.I18n; import eu.cloudnetservice.driver.network.NetworkChannel; @@ -30,6 +31,7 @@ import eu.cloudnetservice.node.impl.cluster.util.QueuedNetworkChannel; import eu.cloudnetservice.node.impl.network.NodeNetworkUtil; import eu.cloudnetservice.node.impl.network.packet.ServiceSyncAckPacket; +import eu.cloudnetservice.node.service.CloudServiceManager; import jakarta.inject.Inject; import jakarta.inject.Singleton; import java.util.Objects; @@ -47,6 +49,7 @@ public final class AuthorizationResponsePacketListener implements PacketListener private final NodeNetworkUtil networkUtil; private final DataSyncRegistry dataSyncRegistry; private final NodeServerProvider nodeServerProvider; + private final CloudServiceManager cloudServiceManager; @Inject public AuthorizationResponsePacketListener( @@ -54,13 +57,15 @@ public AuthorizationResponsePacketListener( @NonNull Configuration configuration, @NonNull NodeNetworkUtil networkUtil, @NonNull DataSyncRegistry dataSyncRegistry, - @NonNull NodeServerProvider nodeServerProvider + @NonNull NodeServerProvider nodeServerProvider, + @NonNull CloudServiceManager cloudServiceManager ) { this.i18n = i18n; this.configuration = configuration; this.networkUtil = networkUtil; this.dataSyncRegistry = dataSyncRegistry; this.nodeServerProvider = nodeServerProvider; + this.cloudServiceManager = cloudServiceManager; } @Override @@ -77,6 +82,7 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { .orElse(null); if (server != null) { var wasReconnect = packetContent.readBoolean(); + var republishServices = false; if (wasReconnect) { try (var syncData = packetContent.readDataBuf()) { var forceApply = syncData.readBoolean(); @@ -99,6 +105,12 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { // of the channel to 'disconnected' before actually closing the channel server.state(NodeServerState.DISCONNECTED); server.channel().close(); + } else if (server.channel() instanceof QueuedNetworkChannel queuedChannel) { + // we reconnected, but the other node either did not notice the disconnect or already removed this node + // from the cluster. flush the packets that were queued for the node and close the old channel + queuedChannel.drainPacketQueue(channel); + queuedChannel.close(); + republishServices = true; } // re-initialize the node data @@ -106,6 +118,22 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { server.state(NodeServerState.READY); channel.packetRegistry().removeListeners(NetworkConstants.INTERNAL_AUTHORIZATION_CHANNEL); this.networkUtil.addDefaultPacketListeners(channel.packetRegistry()); + + // re-select the head node as the node might have been the head node before it disconnected. this + // ensures that the current node uses the same node as the head node as all other nodes in the cluster + this.nodeServerProvider.selectHeadNode(); + + // the other node might have removed all services of this node already, re-publish them to the node + if (republishServices) { + for (var service : this.cloudServiceManager.localCloudServices()) { + ChannelMessage.builder() + .targetNode(server.info().uniqueId()) + .message("update_service_info") + .channel(NetworkConstants.INTERNAL_MSG_CHANNEL) + .build(buffer -> buffer.writeObject(service.serviceInfo())) + .send(); + } + } return; } } diff --git a/node/impl/src/main/java/eu/cloudnetservice/node/impl/service/defaults/DefaultCloudServiceManager.java b/node/impl/src/main/java/eu/cloudnetservice/node/impl/service/defaults/DefaultCloudServiceManager.java index 465e4f658e..f03cd0c7a4 100644 --- a/node/impl/src/main/java/eu/cloudnetservice/node/impl/service/defaults/DefaultCloudServiceManager.java +++ b/node/impl/src/main/java/eu/cloudnetservice/node/impl/service/defaults/DefaultCloudServiceManager.java @@ -162,9 +162,10 @@ public DefaultCloudServiceManager( .dataCollector(this::services) .convertObject(ServiceInfoSnapshot.class) .writer(ser -> { - // ugly hack to get the channel of the service's associated node + // ugly hack to get the channel of the service's associated node. nodes which are currently + // reconnecting are accepted as well, as the services are part of the reconnect data sync var node = this.nodeServerProvider.node(ser.serviceId().nodeUniqueId()); - if (node != null && node.available()) { + if (node != null && (node.available() || node.channel() != null)) { this.handleServiceUpdate(ser, node.channel()); } }) @@ -542,11 +543,17 @@ public void forceRemoveRegisteredService(@NonNull UUID uniqueId) { return null; } - // build the service provider for the newly added service + // build the service provider for the newly added service. the target channel is resolved from the node + // of the service on each call, as the channel changes when the node reconnects to the cluster + var nodeUniqueId = snapshot.serviceId().nodeUniqueId(); var baseRPC = this.sender.invokeMethod("serviceProvider", MTD_SERVICE_PROVIDER, serviceUniqueId); var serviceProvider = this.specificProviderAllocator .withBaseRPC(baseRPC) - .withTargetChannel(() -> source) + .withTargetChannel(() -> { + var node = this.nodeServerProvider.node(nodeUniqueId); + var nodeChannel = node == null ? null : node.channel(); + return nodeChannel == null ? source : nodeChannel; + }) .withAdditionalConstructorParameters(snapshot) .allocate(); diff --git a/node/impl/src/main/resources/lang/de_DE.properties b/node/impl/src/main/resources/lang/de_DE.properties index b075ec3a71..ff53410495 100644 --- a/node/impl/src/main/resources/lang/de_DE.properties +++ b/node/impl/src/main/resources/lang/de_DE.properties @@ -88,6 +88,7 @@ cloudnet-service-post-stop-message-different-node=Der CloudService [uniqueId\={0 # Cluster # cluster-local-db-warning=CloudNet läuft in einem Cluster, aber eine lokal eingebettete Datenbank wird verwendet\! Dies kann große Synchronisationsprobleme verursachen. Bitte überlege zu einer zentralen Datenbank (wie MongoDB) zu wechseln, zu der sich alle Nodes verbinden können. +cluster-server-connection-lost=Die Verbindung zur Node [uniqueId\={0$id$}] wurde unterbrochen\! Es wird bis zu {1$maxDisconnectTime$}ms auf eine erneute Verbindung gewartet cluster-server-soft-disconnect=Node [uniqueId\={0$id$}] wird vorübergehend als getrennt markiert, weil sie zu lange kein Update geschickt hat\! [lastUpdate\={1$time$}ms zurück] cluster-server-hard-disconnect=Die Verbindung zur Node [uniqueId\={0$id$}] wurde geschlossen, da sie sich nicht innerhalt von {1$maxDisconnectTime$}ms erneut verbunden hat\! [lastStateChange\={2$time$}ms zurück] cluster-server-networking-authorization-failed=Autorisierung einer Node mit der Adresse {0$address$} ist leider fehlgeschlagen diff --git a/node/impl/src/main/resources/lang/en_US.properties b/node/impl/src/main/resources/lang/en_US.properties index f62e160f49..1b56fdaef6 100644 --- a/node/impl/src/main/resources/lang/en_US.properties +++ b/node/impl/src/main/resources/lang/en_US.properties @@ -88,6 +88,7 @@ cloudnet-service-post-stop-message-different-node=CloudService [uniqueId={0$id$} # Cluster # cluster-local-db-warning=CloudNet runs in a cluster, but a local embedded database is used! This can cause big synchronization problems. Please consider switching to a central database (like MongoDB) that all nodes can connect to. +cluster-server-connection-lost=Lost the connection to node [uniqueId={0$id$}]! Waiting up to {1$maxDisconnectTime$}ms for the node to reconnect cluster-server-soft-disconnect=Marking node [uniqueId={0$id$}] temporarily as disconnected because it was idling for too long! [lastUpdate={1$time$}ms ago] cluster-server-hard-disconnect=Disconnecting node [uniqueId={0$id$}] because it did not reconnect within {1$maxDisconnectTime$}ms! [lastStateChange={2$time$}ms ago] cluster-server-networking-authorization-failed=Authorization unfortunately failed to the node with the address {0$address$} From bfecae9d5c81a50e6d2b7562c89206496b5b5bec Mon Sep 17 00:00:00 2001 From: Ente_3 Date: Wed, 30 Sep 2026 17:36:41 +0200 Subject: [PATCH 2/2] test: cover node reconnect handling --- .../task/NodeDisconnectTrackerTaskTest.java | 159 ++++++++++++++++ .../util/NodeDisconnectHandlerTest.java | 126 +++++++++++++ .../AuthorizationPacketListenerTest.java | 174 ++++++++++++++++++ ...thorizationResponsePacketListenerTest.java | 121 ++++++++++++ .../listener/RemoteNodeServerTestUtil.java | 79 ++++++++ 5 files changed, 659 insertions(+) create mode 100644 node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTaskTest.java create mode 100644 node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandlerTest.java create mode 100644 node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListenerTest.java create mode 100644 node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListenerTest.java create mode 100644 node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/RemoteNodeServerTestUtil.java diff --git a/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTaskTest.java b/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTaskTest.java new file mode 100644 index 0000000000..6c4111c8b2 --- /dev/null +++ b/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/task/NodeDisconnectTrackerTaskTest.java @@ -0,0 +1,159 @@ +/* + * Copyright 2019-present CloudNetService team & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package eu.cloudnetservice.node.impl.cluster.task; + +import eu.cloudnetservice.driver.cluster.NodeInfoSnapshot; +import eu.cloudnetservice.driver.language.I18n; +import eu.cloudnetservice.node.cluster.LocalNodeServer; +import eu.cloudnetservice.node.cluster.NodeServer; +import eu.cloudnetservice.node.cluster.NodeServerProvider; +import eu.cloudnetservice.node.cluster.NodeServerState; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +public final class NodeDisconnectTrackerTaskTest { + + private static final long GRACE_PERIOD_MILLIS = 30_000; + + private NodeServerProvider provider; + private LocalNodeServer localNode; + private NodeDisconnectTrackerTask trackerTask; + + private static NodeInfoSnapshot mockSnapshot(long startupMillis) { + var snapshot = Mockito.mock(NodeInfoSnapshot.class); + Mockito.when(snapshot.startupMillis()).thenReturn(startupMillis); + return snapshot; + } + + @BeforeEach + void setUp() { + this.provider = Mockito.mock(NodeServerProvider.class); + this.localNode = Mockito.mock(LocalNodeServer.class); + Mockito.when(this.provider.localNode()).thenReturn(this.localNode); + // the local node started after the remote node, so it is responsible for reconnecting + var localSnapshot = mockSnapshot(2_000); + Mockito.when(this.localNode.nodeInfoSnapshot()).thenReturn(localSnapshot); + + var disconnectHandler = Mockito.mock(NodeDisconnectHandler.class); + Mockito.when(disconnectHandler.hardDisconnectMillis()).thenReturn(GRACE_PERIOD_MILLIS); + this.trackerTask = new NodeDisconnectTrackerTask(Mockito.mock(I18n.class), this.provider, disconnectHandler); + } + + private NodeServer registerRemoteNode(NodeServerState state, Instant lastStateChange) { + var server = Mockito.mock(NodeServer.class); + var snapshot = mockSnapshot(1_000); + Mockito.when(server.name()).thenReturn("Node-1"); + Mockito.when(server.state()).thenReturn(state); + Mockito.when(server.lastStateChange()).thenReturn(lastStateChange); + Mockito.when(server.nodeInfoSnapshot()).thenReturn(snapshot); + Mockito.when(server.connect()).thenReturn(CompletableFuture.completedFuture(null)); + Mockito.when(this.provider.nodeServers()).thenReturn(List.of(this.localNode, server)); + return server; + } + + @Test + void testDisconnectedNodeIsReconnectedByYoungerNode() { + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, Instant.now()); + + this.trackerTask.run(); + + Mockito.verify(server).connect(); + Mockito.verify(server, Mockito.never()).close(); + } + + @Test + void testDisconnectedNodeIsNotReconnectedByOlderNode() { + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, Instant.now()); + var localSnapshot = mockSnapshot(500); + Mockito.when(this.localNode.nodeInfoSnapshot()).thenReturn(localSnapshot); + + this.trackerTask.run(); + + Mockito.verify(server, Mockito.never()).connect(); + Mockito.verify(server, Mockito.never()).close(); + } + + @Test + void testReconnectIsNotRepeatedWhileHandshakeIsPending() { + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, Instant.now().minusSeconds(1)); + + this.trackerTask.run(); + this.trackerTask.run(); + + Mockito.verify(server, Mockito.times(1)).connect(); + } + + @Test + void testReconnectIsRepeatedAfterNodeChangedState() { + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, Instant.now().minusSeconds(1)); + this.trackerTask.run(); + + // the node was marked as disconnected again after the last connection was established + Mockito.when(server.lastStateChange()).thenReturn(Instant.now().plusMillis(10)); + this.trackerTask.run(); + + Mockito.verify(server, Mockito.times(2)).connect(); + } + + @Test + void testFailedReconnectIsRepeated() { + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, Instant.now()); + Mockito.when(server.connect()).thenReturn(CompletableFuture.failedFuture(new IllegalStateException("refused"))); + + this.trackerTask.run(); + this.trackerTask.run(); + + Mockito.verify(server, Mockito.times(2)).connect(); + } + + @Test + void testDisconnectedNodeIsRemovedAfterGracePeriod() { + var lastStateChange = Instant.now().minus(Duration.ofMillis(GRACE_PERIOD_MILLIS + 1_000)); + var server = this.registerRemoteNode(NodeServerState.DISCONNECTED, lastStateChange); + + this.trackerTask.run(); + + Mockito.verify(server).close(); + Mockito.verify(server, Mockito.never()).connect(); + } + + @Test + void testSyncingNodeIsRemovedAfterGracePeriod() { + var lastStateChange = Instant.now().minus(Duration.ofMillis(GRACE_PERIOD_MILLIS + 1_000)); + var server = this.registerRemoteNode(NodeServerState.SYNCING, lastStateChange); + + this.trackerTask.run(); + + Mockito.verify(server).close(); + } + + @Test + void testSyncingNodeIsNotReconnected() { + var server = this.registerRemoteNode(NodeServerState.SYNCING, Instant.now()); + + this.trackerTask.run(); + + Mockito.verify(server, Mockito.never()).connect(); + Mockito.verify(server, Mockito.never()).close(); + } +} diff --git a/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandlerTest.java b/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandlerTest.java new file mode 100644 index 0000000000..77089811ba --- /dev/null +++ b/node/impl/src/test/java/eu/cloudnetservice/node/impl/cluster/util/NodeDisconnectHandlerTest.java @@ -0,0 +1,126 @@ +/* + * Copyright 2019-present CloudNetService team & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package eu.cloudnetservice.node.impl.cluster.util; + +import eu.cloudnetservice.driver.event.EventManager; +import eu.cloudnetservice.driver.language.I18n; +import eu.cloudnetservice.driver.network.NetworkChannel; +import eu.cloudnetservice.driver.network.protocol.Packet; +import eu.cloudnetservice.node.cluster.NodeServer; +import eu.cloudnetservice.node.cluster.NodeServerProvider; +import eu.cloudnetservice.node.cluster.NodeServerState; +import eu.cloudnetservice.node.impl.service.InternalCloudServiceManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +public final class NodeDisconnectHandlerTest { + + private static final String GRACE_PERIOD_PROPERTY = "cloudnet.max.node.disconnect.millis"; + + private static NodeDisconnectHandler createHandler(long gracePeriodMillis) { + System.setProperty(GRACE_PERIOD_PROPERTY, Long.toString(gracePeriodMillis)); + return new NodeDisconnectHandler( + Mockito.mock(I18n.class), + Mockito.mock(EventManager.class), + Mockito.mock(InternalCloudServiceManager.class)); + } + + private static NodeServer mockNodeServer(boolean available, boolean head) { + var server = Mockito.mock(NodeServer.class); + Mockito.when(server.name()).thenReturn("Node-2"); + Mockito.when(server.available()).thenReturn(available); + Mockito.when(server.head()).thenReturn(head); + Mockito.when(server.channel()).thenReturn(Mockito.mock(NetworkChannel.class)); + Mockito.when(server.provider()).thenReturn(Mockito.mock(NodeServerProvider.class)); + return server; + } + + @AfterEach + void clearGracePeriod() { + System.clearProperty(GRACE_PERIOD_PROPERTY); + } + + @Test + void testGracePeriodIsReadFromSystemProperty() { + Assertions.assertEquals(0, createHandler(0).hardDisconnectMillis()); + Assertions.assertEquals(30_000, createHandler(30_000).hardDisconnectMillis()); + } + + @Test + void testChannelCloseRemovesNodeWithoutGracePeriod() { + var handler = createHandler(0); + var server = mockNodeServer(true, false); + + handler.handleNodeServerChannelClose(server); + + Mockito.verify(server).close(); + Mockito.verify(server, Mockito.never()).state(Mockito.any()); + } + + @Test + void testChannelCloseMarksNodeDisconnectedWithGracePeriod() { + var handler = createHandler(30_000); + var server = mockNodeServer(true, false); + var closedChannel = server.channel(); + + handler.handleNodeServerChannelClose(server); + + Mockito.verify(server, Mockito.never()).close(); + Mockito.verify(server).state(NodeServerState.DISCONNECTED); + + // packets to the node must be queued instead of being written to the closed channel + var channelCaptor = ArgumentCaptor.forClass(NetworkChannel.class); + Mockito.verify(server).channel(channelCaptor.capture()); + var queuedChannel = Assertions.assertInstanceOf(QueuedNetworkChannel.class, channelCaptor.getValue()); + queuedChannel.sendPacket(Mockito.mock(Packet.class)); + Mockito.verify(closedChannel, Mockito.never()).sendPacket(Mockito.any(Packet.class)); + } + + @Test + void testChannelCloseRemovesUnavailableNodeWithGracePeriod() { + var handler = createHandler(30_000); + var server = mockNodeServer(false, false); + + handler.handleNodeServerChannelClose(server); + + Mockito.verify(server).close(); + Mockito.verify(server, Mockito.never()).state(Mockito.any()); + } + + @Test + void testMarkingHeadNodeDisconnectedSelectsNewHeadNode() { + var handler = createHandler(30_000); + var server = mockNodeServer(true, true); + + handler.markNodeServerDisconnected(server); + + Mockito.verify(server.provider()).selectHeadNode(); + } + + @Test + void testMarkingOtherNodeDisconnectedKeepsHeadNode() { + var handler = createHandler(30_000); + var server = mockNodeServer(true, false); + + handler.markNodeServerDisconnected(server); + + Mockito.verify(server.provider(), Mockito.never()).selectHeadNode(); + } +} diff --git a/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListenerTest.java b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListenerTest.java new file mode 100644 index 0000000000..a40d2193b5 --- /dev/null +++ b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationPacketListenerTest.java @@ -0,0 +1,174 @@ +/* + * Copyright 2019-present CloudNetService team & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package eu.cloudnetservice.node.impl.network.listener; + +import eu.cloudnetservice.driver.cluster.NetworkCluster; +import eu.cloudnetservice.driver.cluster.NetworkClusterNode; +import eu.cloudnetservice.driver.event.EventManager; +import eu.cloudnetservice.driver.impl.network.standard.AuthorizationPacket; +import eu.cloudnetservice.driver.language.I18n; +import eu.cloudnetservice.driver.network.HostAndPort; +import eu.cloudnetservice.driver.network.NetworkChannel; +import eu.cloudnetservice.driver.network.buffer.DataBuf; +import eu.cloudnetservice.driver.network.protocol.Packet; +import eu.cloudnetservice.node.cluster.NodeServerProvider; +import eu.cloudnetservice.node.cluster.NodeServerState; +import eu.cloudnetservice.node.cluster.sync.DataSyncRegistry; +import eu.cloudnetservice.node.config.Configuration; +import eu.cloudnetservice.node.impl.cluster.defaults.RemoteNodeServer; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; +import eu.cloudnetservice.node.impl.cluster.util.QueuedNetworkChannel; +import eu.cloudnetservice.node.impl.junit.EnableServicesInject; +import eu.cloudnetservice.node.impl.network.NodeNetworkUtil; +import eu.cloudnetservice.node.service.CloudServiceManager; +import java.util.List; +import java.util.UUID; +import java.util.function.Predicate; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +@EnableServicesInject +public final class AuthorizationPacketListenerTest { + + private static final UUID CLUSTER_ID = UUID.randomUUID(); + private static final NetworkClusterNode NODE = new NetworkClusterNode( + "Node-1", + List.of(new HostAndPort("127.0.0.1", 1410))); + private static final long NODE_STARTUP_MILLIS = 1_000; + + private NodeServerProvider provider; + private NodeDisconnectHandler disconnectHandler; + private RemoteNodeServer server; + private NetworkChannel oldChannel; + private NetworkChannel newChannel; + private AuthorizationPacketListener listener; + + private static Packet nodeAuthPacket(long startupMillis) { + var content = Mockito.mock(DataBuf.class); + Mockito.when(content.readUniqueId()).thenReturn(CLUSTER_ID); + Mockito.when(content.readObject(NetworkClusterNode.class)).thenReturn(NODE); + // nodes running an older version do not send their startup time + Mockito.when(content.readableBytes()).thenReturn(startupMillis == -1 ? 0 : Long.BYTES); + Mockito.when(content.readLong()).thenReturn(startupMillis); + + var packetContent = Mockito.mock(DataBuf.class); + Mockito.when(packetContent.readObject(AuthorizationPacket.PacketAuthorizationType.class)) + .thenReturn(AuthorizationPacket.PacketAuthorizationType.NODE_TO_NODE); + Mockito.when(packetContent.readDataBuf()).thenReturn(content); + + var packet = Mockito.mock(Packet.class); + Mockito.when(packet.content()).thenReturn(packetContent); + return packet; + } + + @BeforeEach + void setUp() { + this.provider = Mockito.mock(NodeServerProvider.class); + this.disconnectHandler = Mockito.mock(NodeDisconnectHandler.class); + this.server = RemoteNodeServerTestUtil.createRemoteNodeServer(NODE, this.provider, this.disconnectHandler); + Mockito.when(this.provider.nodeServers()).thenReturn(List.of(this.server)); + + this.oldChannel = RemoteNodeServerTestUtil.mockChannel(); + this.newChannel = RemoteNodeServerTestUtil.mockChannel(); + + var configuration = Mockito.mock(Configuration.class); + Mockito.when(configuration.clusterConfig()).thenReturn(new NetworkCluster(CLUSTER_ID, List.of(NODE))); + var dataSyncRegistry = Mockito.mock(DataSyncRegistry.class); + Mockito.when(dataSyncRegistry.prepareClusterData(Mockito.eq(true), Mockito.any(Predicate.class))) + .thenAnswer(_ -> DataBuf.empty()); + + this.listener = new AuthorizationPacketListener( + Mockito.mock(I18n.class), + Mockito.mock(EventManager.class), + configuration, + Mockito.mock(NodeNetworkUtil.class), + dataSyncRegistry, + this.provider, + Mockito.mock(CloudServiceManager.class)); + } + + @Test + void testDisconnectedNodeStartsReconnect() { + var queuedChannel = new QueuedNetworkChannel(this.oldChannel); + RemoteNodeServerTestUtil.initSession( + this.server, queuedChannel, NODE_STARTUP_MILLIS, NodeServerState.DISCONNECTED); + + this.listener.handle(this.newChannel, nodeAuthPacket(NODE_STARTUP_MILLIS)); + + // the old session is kept, the channel gets switched when the sync of the reconnect completes + Assertions.assertEquals(NodeServerState.SYNCING, this.server.state()); + Assertions.assertSame(queuedChannel, this.server.channel()); + Mockito.verify(this.disconnectHandler, Mockito.never()).handleNodeServerClose(Mockito.any()); + } + + @Test + void testDisconnectedNodeOfOlderVersionStartsReconnect() { + RemoteNodeServerTestUtil.initSession( + this.server, new QueuedNetworkChannel(this.oldChannel), NODE_STARTUP_MILLIS, NodeServerState.DISCONNECTED); + + this.listener.handle(this.newChannel, nodeAuthPacket(-1)); + + Assertions.assertEquals(NodeServerState.SYNCING, this.server.state()); + Mockito.verify(this.disconnectHandler, Mockito.never()).handleNodeServerClose(Mockito.any()); + } + + @Test + void testRestartedDisconnectedNodeJoinsAsNewNode() { + RemoteNodeServerTestUtil.initSession( + this.server, new QueuedNetworkChannel(this.oldChannel), NODE_STARTUP_MILLIS, NodeServerState.DISCONNECTED); + + this.listener.handle(this.newChannel, nodeAuthPacket(NODE_STARTUP_MILLIS + 5_000)); + + // the previous session gets closed and the node is accepted like a newly started node + Mockito.verify(this.disconnectHandler).handleNodeServerClose(this.server); + Mockito.verify(this.oldChannel).close(); + Assertions.assertEquals(NodeServerState.READY, this.server.state()); + Assertions.assertSame(this.newChannel, this.server.channel()); + Assertions.assertNull(this.server.nodeInfoSnapshot()); + } + + @Test + void testRestartedReadyNodeJoinsAsNewNode() { + RemoteNodeServerTestUtil.initSession(this.server, this.oldChannel, NODE_STARTUP_MILLIS, NodeServerState.READY); + + this.listener.handle(this.newChannel, nodeAuthPacket(NODE_STARTUP_MILLIS + 5_000)); + + Mockito.verify(this.disconnectHandler).handleNodeServerClose(this.server); + Assertions.assertEquals(NodeServerState.READY, this.server.state()); + Assertions.assertSame(this.newChannel, this.server.channel()); + } + + @Test + void testInterruptedReconnectFlushesQueuedPackets() { + var queuedPacket = Mockito.mock(Packet.class); + var queuedChannel = new QueuedNetworkChannel(this.oldChannel); + queuedChannel.sendPacket(queuedPacket); + RemoteNodeServerTestUtil.initSession(this.server, queuedChannel, NODE_STARTUP_MILLIS, NodeServerState.SYNCING); + Mockito.clearInvocations(this.provider); + + this.listener.handle(this.newChannel, nodeAuthPacket(NODE_STARTUP_MILLIS)); + + Mockito.verify(this.newChannel).sendPacketSync(queuedPacket); + Mockito.verify(this.oldChannel).close(); + Mockito.verify(this.provider).selectHeadNode(); + Mockito.verify(this.disconnectHandler, Mockito.never()).handleNodeServerClose(Mockito.any()); + Assertions.assertEquals(NodeServerState.READY, this.server.state()); + Assertions.assertSame(this.newChannel, this.server.channel()); + } +} diff --git a/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListenerTest.java b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListenerTest.java new file mode 100644 index 0000000000..8c4fdc9357 --- /dev/null +++ b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/AuthorizationResponsePacketListenerTest.java @@ -0,0 +1,121 @@ +/* + * Copyright 2019-present CloudNetService team & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package eu.cloudnetservice.node.impl.network.listener; + +import eu.cloudnetservice.driver.cluster.NetworkCluster; +import eu.cloudnetservice.driver.cluster.NetworkClusterNode; +import eu.cloudnetservice.driver.language.I18n; +import eu.cloudnetservice.driver.network.HostAndPort; +import eu.cloudnetservice.driver.network.NetworkChannel; +import eu.cloudnetservice.driver.network.buffer.DataBuf; +import eu.cloudnetservice.driver.network.protocol.Packet; +import eu.cloudnetservice.node.cluster.NodeServerProvider; +import eu.cloudnetservice.node.cluster.NodeServerState; +import eu.cloudnetservice.node.cluster.sync.DataSyncRegistry; +import eu.cloudnetservice.node.config.Configuration; +import eu.cloudnetservice.node.impl.cluster.defaults.RemoteNodeServer; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; +import eu.cloudnetservice.node.impl.cluster.util.QueuedNetworkChannel; +import eu.cloudnetservice.node.impl.junit.EnableServicesInject; +import eu.cloudnetservice.node.impl.network.NodeNetworkUtil; +import eu.cloudnetservice.node.service.CloudServiceManager; +import java.util.List; +import java.util.UUID; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +@EnableServicesInject +public final class AuthorizationResponsePacketListenerTest { + + private static final HostAndPort NODE_LISTENER = new HostAndPort("127.0.0.1", 1410); + private static final NetworkClusterNode NODE = new NetworkClusterNode("Node-1", List.of(NODE_LISTENER)); + + private NodeServerProvider provider; + private CloudServiceManager cloudServiceManager; + private RemoteNodeServer server; + private NetworkChannel oldChannel; + private NetworkChannel newChannel; + private AuthorizationResponsePacketListener listener; + + private static Packet plainAuthSuccessPacket() { + var content = Mockito.mock(DataBuf.class); + // the authorization was successful, but the other node did not handle it as a reconnect + Mockito.when(content.readBoolean()).thenReturn(true, false); + var packet = Mockito.mock(Packet.class); + Mockito.when(packet.content()).thenReturn(content); + return packet; + } + + @BeforeEach + void setUp() { + this.provider = Mockito.mock(NodeServerProvider.class); + this.server = RemoteNodeServerTestUtil.createRemoteNodeServer( + NODE, + this.provider, + Mockito.mock(NodeDisconnectHandler.class)); + Mockito.when(this.provider.node(NODE.uniqueId())).thenReturn(this.server); + + this.oldChannel = RemoteNodeServerTestUtil.mockChannel(); + this.newChannel = RemoteNodeServerTestUtil.mockChannel(); + Mockito.when(this.newChannel.serverAddress()).thenReturn(NODE_LISTENER); + + var configuration = Mockito.mock(Configuration.class); + Mockito.when(configuration.clusterConfig()).thenReturn(new NetworkCluster(UUID.randomUUID(), List.of(NODE))); + this.cloudServiceManager = Mockito.mock(CloudServiceManager.class); + Mockito.when(this.cloudServiceManager.localCloudServices()).thenReturn(List.of()); + + this.listener = new AuthorizationResponsePacketListener( + Mockito.mock(I18n.class), + configuration, + Mockito.mock(NodeNetworkUtil.class), + Mockito.mock(DataSyncRegistry.class), + this.provider, + this.cloudServiceManager); + } + + @Test + void testNewConnectionSelectsHeadNode() { + this.listener.handle(this.newChannel, plainAuthSuccessPacket()); + + Assertions.assertEquals(NodeServerState.READY, this.server.state()); + Assertions.assertSame(this.newChannel, this.server.channel()); + Mockito.verify(this.provider).selectHeadNode(); + // this node never lost a connection to the node, there is nothing to re-publish + Mockito.verify(this.cloudServiceManager, Mockito.never()).localCloudServices(); + } + + @Test + void testPlainAuthAfterReconnectFlushesQueuedPackets() { + var queuedPacket = Mockito.mock(Packet.class); + var queuedChannel = new QueuedNetworkChannel(this.oldChannel); + queuedChannel.sendPacket(queuedPacket); + RemoteNodeServerTestUtil.initSession(this.server, queuedChannel, 1_000, NodeServerState.DISCONNECTED); + Mockito.clearInvocations(this.provider); + + this.listener.handle(this.newChannel, plainAuthSuccessPacket()); + + Mockito.verify(this.newChannel).sendPacketSync(queuedPacket); + Mockito.verify(this.oldChannel).close(); + Mockito.verify(this.provider).selectHeadNode(); + // the other node might have removed the services of this node, they must be re-published + Mockito.verify(this.cloudServiceManager).localCloudServices(); + Assertions.assertEquals(NodeServerState.READY, this.server.state()); + Assertions.assertSame(this.newChannel, this.server.channel()); + } +} diff --git a/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/RemoteNodeServerTestUtil.java b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/RemoteNodeServerTestUtil.java new file mode 100644 index 0000000000..7c062fb8d1 --- /dev/null +++ b/node/impl/src/test/java/eu/cloudnetservice/node/impl/network/listener/RemoteNodeServerTestUtil.java @@ -0,0 +1,79 @@ +/* + * Copyright 2019-present CloudNetService team & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package eu.cloudnetservice.node.impl.network.listener; + +import eu.cloudnetservice.driver.cluster.NetworkClusterNode; +import eu.cloudnetservice.driver.cluster.NodeInfoSnapshot; +import eu.cloudnetservice.driver.network.NetworkChannel; +import eu.cloudnetservice.driver.network.NetworkClient; +import eu.cloudnetservice.driver.network.protocol.PacketListenerRegistry; +import eu.cloudnetservice.driver.network.rpc.factory.RPCImplementationBuilder; +import eu.cloudnetservice.driver.provider.CloudServiceFactory; +import eu.cloudnetservice.driver.provider.CloudServiceProvider; +import eu.cloudnetservice.node.cluster.NodeServerProvider; +import eu.cloudnetservice.node.cluster.NodeServerState; +import eu.cloudnetservice.node.cluster.sync.DataSyncRegistry; +import eu.cloudnetservice.node.impl.cluster.defaults.RemoteNodeServer; +import eu.cloudnetservice.node.impl.cluster.util.NodeDisconnectHandler; +import lombok.NonNull; +import org.mockito.Mockito; + +final class RemoteNodeServerTestUtil { + + private RemoteNodeServerTestUtil() { + throw new UnsupportedOperationException(); + } + + @SuppressWarnings("unchecked") + static @NonNull RemoteNodeServer createRemoteNodeServer( + @NonNull NetworkClusterNode node, + @NonNull NodeServerProvider provider, + @NonNull NodeDisconnectHandler disconnectHandler + ) { + var allocator = (RPCImplementationBuilder.InstanceAllocator) Mockito.mock( + RPCImplementationBuilder.InstanceAllocator.class); + Mockito.when(allocator.withTargetChannel(Mockito.any())).thenReturn(allocator); + Mockito.when(allocator.allocate()).thenReturn(Mockito.mock(CloudServiceFactory.class)); + return new RemoteNodeServer( + node, + provider, + Mockito.mock(NetworkClient.class), + Mockito.mock(DataSyncRegistry.class), + disconnectHandler, + Mockito.mock(CloudServiceProvider.class), + allocator); + } + + static void initSession( + @NonNull RemoteNodeServer server, + @NonNull NetworkChannel channel, + long startupMillis, + @NonNull NodeServerState state + ) { + var snapshot = Mockito.mock(NodeInfoSnapshot.class); + Mockito.when(snapshot.startupMillis()).thenReturn(startupMillis); + server.channel(channel); + server.updateNodeInfoSnapshot(snapshot); + server.state(state); + } + + static @NonNull NetworkChannel mockChannel() { + var channel = Mockito.mock(NetworkChannel.class); + Mockito.when(channel.packetRegistry()).thenReturn(Mockito.mock(PacketListenerRegistry.class)); + return channel; + } +}