feat(20242): Improve Reconnect orchestration (#21599)

Signed-off-by: mxtartaglia <[email protected]>

Author: mxtartaglia-sl

platform-sdk/consensus-gossip/src/main/java/module-info.java

@@ -2,6 +2,5 @@
 module org.hiero.consensus.gossip {
     exports org.hiero.consensus.gossip;
 
-    requires transitive org.hiero.consensus.model;
     requires static transitive com.github.spotbugs.annotations;
 }

platform-sdk/consensus-gossip/src/main/java/org/hiero/consensus/gossip/FallenBehindManager.java

@@ -127,6 +127,8 @@ void randomDataTest(final int networkSize) {
                             mock(BindableInputWire.class),
                             mock(BindableInputWire.class),
                             mock(BindableInputWire.class),
+                            mock(BindableInputWire.class),
+                            mock(BindableInputWire.class),
                             mock(StandardOutputWire.class));
             model.start();
         }

platform-sdk/consensus-otter-tests/src/test/java/org/hiero/otter/fixtures/turtle/gossip/SimulatedGossipTests.java

@@ -5,6 +5,7 @@
 import com.hedera.hapi.node.state.roster.Roster;
 import com.swirlds.common.test.fixtures.WeightGenerator;
 import com.swirlds.common.test.fixtures.WeightGenerators;
+import com.swirlds.platform.reconnect.FallenBehindStatus;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.nio.file.Path;
@@ -453,7 +454,7 @@ default Partition createNetworkPartition(@NonNull final Node node0, @NonNull fin
      *
      * @param maybeBehindNode the node to check behind status for
      * @return {@code true} if the node is behind by node weight, {@code false} otherwise
-     * @see com.swirlds.platform.gossip.shadowgraph.SyncFallenBehindStatus
+     * @see FallenBehindStatus
      */
     boolean nodeIsBehindByNodeWeight(@NonNull Node maybeBehindNode);
 
@@ -478,7 +479,7 @@ default boolean nodeIsBehindByNodeCount(@NonNull Node maybeBehindNode) {
      * @param maybeBehindNode the node to check behind status for
      * @param otherNodes additional nodes to consider for the behind check (optional)
      * @return {@code true} if the node is behind by the specified fraction of peers, {@code false} otherwise
-     * @see com.swirlds.platform.gossip.shadowgraph.SyncFallenBehindStatus
+     * @see FallenBehindStatus
      */
     boolean nodesAreBehindByNodeCount(@NonNull Node maybeBehindNode, @Nullable Node... otherNodes);
 

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/Network.java

@@ -18,7 +18,7 @@
 import com.swirlds.common.test.fixtures.WeightGenerators;
 import com.swirlds.common.utility.Threshold;
 import com.swirlds.platform.crypto.CryptoStatic;
-import com.swirlds.platform.gossip.shadowgraph.SyncFallenBehindStatus;
+import com.swirlds.platform.reconnect.FallenBehindStatus;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.nio.file.Path;
@@ -824,8 +824,8 @@ public boolean nodeIsBehindByNodeWeight(@NonNull final Node maybeBehindNode) {
                     maybeAheadNode.newConsensusResult().getLatestEventWindow();
 
             // If any peer in the required list says the "self" node is not behind, the node is not behind.
-            if (SyncFallenBehindStatus.getStatus(selfEventWindow, peerEventWindow)
-                    != SyncFallenBehindStatus.SELF_FALLEN_BEHIND) {
+            if (FallenBehindStatus.getStatus(selfEventWindow, peerEventWindow)
+                    != FallenBehindStatus.SELF_FALLEN_BEHIND) {
                 weightOfAheadNodes += maybeAheadNode.weight();
             }
         }
@@ -861,8 +861,8 @@ public boolean nodesAreBehindByNodeCount(
                                         - 5)); // add buffer to account for unpropagated event windows
 
                 // If any peer in the required list says the "self" node is behind, it is ahead so add it to the count
-                if (SyncFallenBehindStatus.getStatus(selfEventWindow, peerEventWindowWithBuffer)
-                        == SyncFallenBehindStatus.SELF_FALLEN_BEHIND) {
+                if (FallenBehindStatus.getStatus(selfEventWindow, peerEventWindowWithBuffer)
+                        == FallenBehindStatus.SELF_FALLEN_BEHIND) {
                     numNodesAhead++;
                 }
             }

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/internal/AbstractNetwork.java

@@ -74,6 +74,8 @@ public void bind(
             @NonNull final BindableInputWire<NoInput, Void> startInput,
             @NonNull final BindableInputWire<NoInput, Void> stopInput,
             @NonNull final BindableInputWire<NoInput, Void> clearInput,
+            @NonNull final BindableInputWire<NoInput, Void> pauseInput,
+            @NonNull final BindableInputWire<NoInput, Void> resumeInput,
             @NonNull final BindableInputWire<Duration, Void> systemHealthInput,
             @NonNull final BindableInputWire<PlatformStatus, Void> platformStatusInput,
             @NonNull final StandardOutputWire<Double> syncLagOutput) {
@@ -86,6 +88,8 @@ public void bind(
         startInput.bindConsumer(ignored -> {});
         stopInput.bindConsumer(ignored -> {});
         clearInput.bindConsumer(ignored -> {});
+        pauseInput.bindConsumer(ignored -> {});
+        resumeInput.bindConsumer(ignored -> {});
         systemHealthInput.bindConsumer(ignored -> {});
         platformStatusInput.bindConsumer(ignored -> {});
     }

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/turtle/gossip/SimulatedGossip.java

@@ -13,6 +13,7 @@
 import static org.hiero.otter.fixtures.assertions.StatusProgressionStep.target;
 
 import com.swirlds.common.merkle.synchronization.config.ReconnectConfig_;
+import com.swirlds.logging.legacy.payload.ReconnectStartPayload;
 import com.swirlds.platform.consensus.ConsensusConfig_;
 import com.swirlds.platform.wiring.PlatformSchedulersConfig_;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -31,7 +32,6 @@
 import org.hiero.otter.fixtures.result.MultipleNodePlatformStatusResults;
 import org.hiero.otter.fixtures.result.SingleNodePlatformStatusResult;
 import org.hiero.otter.fixtures.result.SubscriberAction;
-import org.junit.jupiter.api.Disabled;
 
 /**
  * Tests the reconnect functionality of a node that has fallen behind in the consensus rounds. The test ensures that the
@@ -42,6 +42,8 @@ public class ReconnectTest {
     /** Reducing the number of rounds non-expired will allow nodes to require a reconnect faster. */
     private static final long ROUNDS_EXPIRED = 100L;
 
+    public static final Duration BEHIND_WAIT_TIME = Duration.ofSeconds(240L);
+
     /**
      * Tests that a node which is killed, kept down until it is behind, and then restarted is able to reconnect to the
      * network and catch up with consensus.
@@ -249,7 +251,6 @@ private void disableSyntheticBottleneck(@NonNull final Node... nodesToThrottle)
      * @param env the test environment
      */
     @OtterTest(requires = Capability.RECONNECT)
-    @Disabled
     void testReconnectSucceedsAfterFailure(@NonNull final TestEnvironment env) {
         final Network network = env.network();
         final TimeManager timeManager = env.timeManager();
@@ -263,7 +264,9 @@ void testReconnectSucceedsAfterFailure(@NonNull final TestEnvironment env) {
                         PlatformSchedulersConfig_.TRANSACTION_HANDLER,
                         "SEQUENTIAL_THREAD CAPACITY(100) FLUSHABLE SQUELCHABLE")
                 .withConfigValue(ConsensusConfig_.ROUNDS_EXPIRED, ROUNDS_EXPIRED)
-                .withConfigValue(ReconnectConfig_.ASYNC_STREAM_TIMEOUT, Duration.ofSeconds(1));
+                .withConfigValue(ReconnectConfig_.ASYNC_STREAM_TIMEOUT, Duration.ofSeconds(1))
+                .withConfigValue(ReconnectConfig_.MAXIMUM_RECONNECT_FAILURES_BEFORE_SHUTDOWN, 2)
+                .withConfigValue(ReconnectConfig_.MINIMUM_TIME_BETWEEN_RECONNECTS, Duration.ofMillis(10));
 
         network.start();
 
@@ -280,7 +283,7 @@ void testReconnectSucceedsAfterFailure(@NonNull final TestEnvironment env) {
         enableSyntheticBottleneck(Duration.ofMinutes(10), nodeToReconnect);
         timeManager.waitForCondition(
                 nodeToReconnect::isBehind,
-                Duration.ofSeconds(120L),
+                BEHIND_WAIT_TIME,
                 "Node did not enter BEHIND status within the expected time "
                         + "frame after synthetic bottleneck was enabled");
 
@@ -308,7 +311,6 @@ void testReconnectSucceedsAfterFailure(@NonNull final TestEnvironment env) {
      * @param env the test environment
      */
     @OtterTest(requires = {Capability.RECONNECT, Capability.SINGLE_NODE_JVM_SHUTDOWN})
-    @Disabled
     void testNodeShutsDownAfterMaxFailedReconnects(@NonNull final TestEnvironment env) {
         final Network network = env.network();
         final TimeManager timeManager = env.timeManager();
@@ -325,24 +327,29 @@ void testNodeShutsDownAfterMaxFailedReconnects(@NonNull final TestEnvironment en
                         "SEQUENTIAL_THREAD CAPACITY(100) FLUSHABLE SQUELCHABLE")
                 .withConfigValue(ConsensusConfig_.ROUNDS_EXPIRED, ROUNDS_EXPIRED)
                 .withConfigValue(ReconnectConfig_.ASYNC_STREAM_TIMEOUT, Duration.ofSeconds(1))
-                .withConfigValue(ReconnectConfig_.MAXIMUM_RECONNECT_FAILURES_BEFORE_SHUTDOWN, maxFailedReconnects);
+                .withConfigValue(ReconnectConfig_.MAXIMUM_RECONNECT_FAILURES_BEFORE_SHUTDOWN, maxFailedReconnects)
+                .withConfigValue(ReconnectConfig_.MINIMUM_TIME_BETWEEN_RECONNECTS, Duration.ofMillis(10));
 
         network.start();
 
         final Node nodeToReconnect = network.nodes().getLast();
 
-        nodeToReconnect.newLogResult().subscribe(logEntry -> {
-            if (logEntry.message().contains("Starting reconnect in role of the receiver.")) {
-                network.isolate(nodeToReconnect);
-                return SubscriberAction.UNSUBSCRIBE;
+        network.newReconnectResults().subscribe(notification -> {
+            if (notification.payload().getClass().equals(ReconnectStartPayload.class)) {
+                final ReconnectStartPayload payload = (ReconnectStartPayload) notification.payload();
+                final Node node = network.nodes().stream()
+                        .filter(n -> n.selfId().id() == payload.getOtherNodeId())
+                        .findFirst()
+                        .orElse(nodeToReconnect);
+                network.isolate(node);
             }
             return SubscriberAction.CONTINUE;
         });
 
         enableSyntheticBottleneck(Duration.ofMinutes(10), nodeToReconnect);
         timeManager.waitForCondition(
                 nodeToReconnect::isBehind,
-                Duration.ofSeconds(120L),
+                BEHIND_WAIT_TIME,
                 "Node did not enter BEHIND status within the expected time "
                         + "frame after synthetic bottleneck was enabled");
 
@@ -359,4 +366,59 @@ void testNodeShutsDownAfterMaxFailedReconnects(@NonNull final TestEnvironment en
                 Duration.ofMinutes(2L),
                 "Node did not shut down within the expected time frame after exceeding the maximum number of failed reconnects.");
     }
+
+    @OtterTest(requires = {Capability.RECONNECT, Capability.SINGLE_NODE_JVM_SHUTDOWN})
+    void testIsolateNodeWhileReconnectingAndRestore(@NonNull final TestEnvironment env) {
+        final Network network = env.network();
+        final TimeManager timeManager = env.timeManager();
+
+        network.addNodes(5);
+
+        final int maxFailedReconnects = 3;
+
+        // For this test to work, we need to lower the limit for the transaction handler component
+        // With the new limit set, once the transaction handler has 100 pending transactions, the node will stop
+        // gossipping and stop creating events. This will cause the node to go into the checking state.
+        network.withConfigValue(
+                        PlatformSchedulersConfig_.TRANSACTION_HANDLER,
+                        "SEQUENTIAL_THREAD CAPACITY(100) FLUSHABLE SQUELCHABLE")
+                .withConfigValue(ConsensusConfig_.ROUNDS_EXPIRED, ROUNDS_EXPIRED)
+                .withConfigValue(ReconnectConfig_.ASYNC_STREAM_TIMEOUT, Duration.ofSeconds(1))
+                .withConfigValue(ReconnectConfig_.MAXIMUM_RECONNECT_FAILURES_BEFORE_SHUTDOWN, maxFailedReconnects)
+                .withConfigValue(ReconnectConfig_.MINIMUM_TIME_BETWEEN_RECONNECTS, Duration.ofMillis(10));
+
+        network.start();
+
+        final Node nodeToReconnect = network.nodes().getLast();
+
+        network.newReconnectResults().subscribe(notification -> {
+            if (notification.payload().getClass().equals(ReconnectStartPayload.class)) {
+                network.isolate(nodeToReconnect);
+                return SubscriberAction.UNSUBSCRIBE;
+            }
+            return SubscriberAction.CONTINUE;
+        });
+
+        enableSyntheticBottleneck(Duration.ofMinutes(10), nodeToReconnect);
+        timeManager.waitForCondition(
+                nodeToReconnect::isBehind,
+                BEHIND_WAIT_TIME,
+                "Node did not enter BEHIND status within the expected time "
+                        + "frame after synthetic bottleneck was enabled");
+
+        network.setBandwidthForAllConnections(nodeToReconnect, BandwidthLimit.ofKilobytesPerSecond(1));
+
+        disableSyntheticBottleneck(nodeToReconnect);
+
+        timeManager.waitForCondition(
+                () -> nodeToReconnect.newReconnectResult().numFailedReconnects() > 0,
+                Duration.ofMinutes(2L),
+                "Node did not record the expected number of failed reconnect attempts within the expected time frame.");
+        timeManager.waitFor(Duration.ofSeconds(30));
+        network.restoreConnectivity();
+        timeManager.waitForCondition(
+                nodeToReconnect::isActive,
+                Duration.ofMinutes(2L),
+                "Node did not become ACTIVE within the expected time frame after restoring normal connectivity");
+    }
 }

platform-sdk/consensus-otter-tests/src/testOtter/java/org/hiero/otter/test/ReconnectTest.java

@@ -7,7 +7,6 @@ public class ReconnectFinishPayload extends AbstractLogPayload {
     private long nodeId;
     private long otherNodeId;
     private long round;
-    private boolean success;
 
     public ReconnectFinishPayload() {}
 
@@ -68,12 +67,4 @@ public long getRound() {
     public void setRound(long round) {
         this.round = round;
     }
-
-    public boolean isSuccess() {
-        return success;
-    }
-
-    public void setSuccess(final boolean success) {
-        this.success = success;
-    }
 }