Fix/reconnect multinode timeout

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

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<String, Instant> 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,44 +76,63 @@ 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));
         }
       }
 
       // 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);
           }
         }
       }
     } catch (Exception exception) {
       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());
+    }
+  }
 }

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<CloudService> 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())) {

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,20 +50,23 @@ 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(
     @NonNull @Service I18n i18n,
     @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);
     }
   }
 }

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,13 +62,15 @@ public DefaultNetworkServerChannelHandler(
     @NonNull NodeNetworkUtil networkUtil,
     @NonNull Configuration configuration,
     @NonNull NodeServerProvider nodeServerProvider,
+    @NonNull NodeDisconnectHandler disconnectHandler,
     @NonNull CloudServiceManager cloudServiceManager
   ) {
     this.i18n = i18n;
     this.eventManager = eventManager;
     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);
     }
   }
 

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,13 +89,21 @@ 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;
           }
           // 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));
               }