Merge pull request #46 from Consensys/fix-sync-for-old-state

fix sync for old state

Author: matkt

CHANGELOG.md

@@ -1,8 +1,11 @@
 # Changelog
 
 ## Unreleased
+- Fixed a bug that prevented the node from syncing with an old snapshot [#46](https://github.com/Consensys/besu-fleet-plugin/pull/46)
+## 0.3.8
 - `--plugin-fleet-follower-http-host` and `--plugin-fleet-follower-http-port` options are deprecated and will be removed in a future release.
 - use BesuConfiguration service to get RPC host and port values as configured in Besu
+- Changes the logic to have the new head information directly in the besu event. [#45](https://github.com/Consensys/besu-fleet-plugin/pull/45)
 
 ## 0.2.0
 - fix cli parameter parsing regression in 0.1.0

src/main/java/net/consensys/fleet/common/plugin/FleetPlugin.java

@@ -33,7 +33,8 @@
 import net.consensys.fleet.leader.peer.LeaderPeerNetworkMaintainer;
 import net.consensys.fleet.leader.rpc.client.FleetShipNewHeadClient;
 import net.consensys.fleet.leader.rpc.server.FleetAddFollowerServer;
-import net.consensys.fleet.leader.rpc.server.FleetGetBlockServer;
+import net.consensys.fleet.leader.rpc.server.FleetGetBlockByHashServer;
+import net.consensys.fleet.leader.rpc.server.FleetGetBlockByNumberServer;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -237,7 +238,8 @@ private List<PluginRpcMethod> createServerMethods() {
         new BlockContextProvider(pluginServiceProvider, new FleetGetBlockClient(webClient));
     methods.add(new FleetGetConfigServer(convertMapperProvider));
     methods.add(new FleetAddFollowerServer(peerManagers));
-    methods.add(new FleetGetBlockServer(convertMapperProvider, pluginServiceProvider));
+    methods.add(new FleetGetBlockByHashServer(convertMapperProvider, pluginServiceProvider));
+    methods.add(new FleetGetBlockByNumberServer(convertMapperProvider, pluginServiceProvider));
     fleetModeSynchronizer =
         new FleetModeSynchronizer(
             pluginServiceProvider,

src/main/java/net/consensys/fleet/common/rpc/client/AbstractStateRpcSender.java

@@ -25,8 +25,6 @@ public AbstractStateRpcSender(final WebClientWrapper webClient) {
     this.webClient = webClient;
   }
 
-  protected abstract String getMethodeName();
-
   public static String getENDPOINT() {
     return ENDPOINT;
   }

src/main/java/net/consensys/fleet/common/rpc/model/GetBlockRequest.java renamed to src/main/java/net/consensys/fleet/common/rpc/model/AbstractGetBlockRequest.java

@@ -15,28 +15,21 @@
 package net.consensys.fleet.common.rpc.model;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import org.hyperledger.besu.datatypes.Hash;
 
-public class GetBlockRequest {
-
-  @JsonProperty("blockHash")
-  private Hash blockHash;
+public abstract class AbstractGetBlockRequest {
 
   @JsonProperty("fetchReceipts")
   private boolean fetchReceipts;
 
-  public GetBlockRequest() {}
+  public AbstractGetBlockRequest() {}
 
-  public GetBlockRequest(final Hash blockHash, final boolean fetchReceipts) {
-    this.blockHash = blockHash;
+  public AbstractGetBlockRequest(final boolean fetchReceipts) {
     this.fetchReceipts = fetchReceipts;
   }
 
-  public Hash getBlockHash() {
-    return blockHash;
-  }
-
   public boolean isFetchReceipts() {
     return fetchReceipts;
   }
+
+  public abstract String getMethodName();
 }

src/main/java/net/consensys/fleet/common/rpc/model/GetBlockByHashRequest.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright ConsenSys 2023
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package net.consensys.fleet.common.rpc.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.hyperledger.besu.datatypes.Hash;
+
+public class GetBlockByHashRequest extends AbstractGetBlockRequest {
+
+  @JsonProperty("blockHash")
+  private Hash blockHash;
+
+  public GetBlockByHashRequest() {}
+
+  public GetBlockByHashRequest(final Hash blockHash, final boolean fetchReceipts) {
+    super(fetchReceipts);
+    this.blockHash = blockHash;
+  }
+
+  @JsonIgnore
+  @Override
+  public String getMethodName() {
+    return "fleet_getBlockByHash";
+  }
+
+  public Hash getBlockHash() {
+    return blockHash;
+  }
+}

src/main/java/net/consensys/fleet/common/rpc/model/GetBlockByNumberRequest.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright ConsenSys 2023
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package net.consensys.fleet.common.rpc.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class GetBlockByNumberRequest extends AbstractGetBlockRequest {
+
+  @JsonProperty("blockNumber")
+  private long blockNumber;
+
+  public GetBlockByNumberRequest() {}
+
+  public GetBlockByNumberRequest(final long blockNumber, final boolean fetchReceipts) {
+    super(fetchReceipts);
+    this.blockNumber = blockNumber;
+  }
+
+  @JsonIgnore
+  @Override
+  public String getMethodName() {
+    return "fleet_getBlockByNumber";
+  }
+
+  public long getBlockNumber() {
+    return blockNumber;
+  }
+}

src/main/java/net/consensys/fleet/follower/rpc/client/FleetAddFollowerClient.java

@@ -32,17 +32,12 @@ public FleetAddFollowerClient(final WebClientWrapper webClient) {
     super(webClient);
   }
 
-  @Override
-  protected String getMethodeName() {
-    return METHOD_NAME;
-  }
-
   @Override
   public CompletableFuture<Boolean> sendData(PeerNode data) {
     final CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
     try {
       webClient
-          .sendToLeader(ENDPOINT, getMethodeName(), data)
+          .sendToLeader(ENDPOINT, METHOD_NAME, data)
           .whenComplete(
               (bufferHttpResponse, throwable) -> {
                 completableFuture.complete(isConnected(bufferHttpResponse, throwable));

src/main/java/net/consensys/fleet/follower/rpc/client/FleetGetBlockClient.java

@@ -16,32 +16,27 @@
 
 import net.consensys.fleet.common.rpc.client.AbstractStateRpcSender;
 import net.consensys.fleet.common.rpc.client.WebClientWrapper;
-import net.consensys.fleet.common.rpc.model.GetBlockRequest;
+import net.consensys.fleet.common.rpc.model.AbstractGetBlockRequest;
 import net.consensys.fleet.common.rpc.model.GetBlockResponse;
 
 import java.util.concurrent.CompletableFuture;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 
-public class FleetGetBlockClient extends AbstractStateRpcSender<GetBlockRequest, GetBlockResponse> {
-
-  private static final String METHOD_NAME = "fleet_getBlock";
+public class FleetGetBlockClient
+    extends AbstractStateRpcSender<AbstractGetBlockRequest, GetBlockResponse> {
 
   public FleetGetBlockClient(final WebClientWrapper webClient) {
     super(webClient);
   }
 
   @Override
-  protected String getMethodeName() {
-    return METHOD_NAME;
-  }
-
-  @Override
-  public CompletableFuture<GetBlockResponse> sendData(final GetBlockRequest blockNumber) {
+  public CompletableFuture<GetBlockResponse> sendData(
+      final AbstractGetBlockRequest getBlockRequest) {
     final CompletableFuture<GetBlockResponse> completableFuture = new CompletableFuture<>();
     try {
       webClient
-          .sendToLeader(ENDPOINT, getMethodeName(), blockNumber)
+          .sendToLeader(ENDPOINT, getBlockRequest.getMethodName(), getBlockRequest)
           .whenCompleteAsync(
               (result, throwable) -> {
                 if (throwable == null) {

src/main/java/net/consensys/fleet/follower/sync/BlockContextProvider.java

@@ -15,7 +15,9 @@
 package net.consensys.fleet.follower.sync;
 
 import net.consensys.fleet.common.plugin.PluginServiceProvider;
-import net.consensys.fleet.common.rpc.model.GetBlockRequest;
+import net.consensys.fleet.common.rpc.model.AbstractGetBlockRequest;
+import net.consensys.fleet.common.rpc.model.GetBlockByHashRequest;
+import net.consensys.fleet.common.rpc.model.GetBlockByNumberRequest;
 import net.consensys.fleet.common.rpc.model.GetBlockResponse;
 import net.consensys.fleet.common.rpc.model.NewHeadParams;
 import net.consensys.fleet.follower.rpc.client.FleetGetBlockClient;
@@ -53,8 +55,8 @@ public BlockContextProvider(
     this.getBlockClient = getBlockClient;
   }
 
-  public Optional<FleetBlockContext> getLeaderBlockContextByNumber(
-      final CompositeBlockKey compositeBlockKey, final boolean fetchReceipts) {
+  private Optional<FleetBlockContext> getLeaderBlockContext(
+      final CompositeBlockKey compositeBlockKey, AbstractGetBlockRequest request) {
     try {
       Optional<FleetBlockContext> cachedContext =
           Optional.ofNullable(leaderBlock.getIfPresent(compositeBlockKey));
@@ -63,10 +65,7 @@ public Optional<FleetBlockContext> getLeaderBlockContextByNumber(
         return cachedContext;
       }
 
-      GetBlockResponse response =
-          getBlockClient
-              .sendData(new GetBlockRequest(compositeBlockKey.getBlockHash(), fetchReceipts))
-              .get();
+      GetBlockResponse response = getBlockClient.sendData(request).get();
 
       FleetBlockContext context =
           new FleetBlockContext(
@@ -82,6 +81,20 @@ public Optional<FleetBlockContext> getLeaderBlockContextByNumber(
     }
   }
 
+  public Optional<FleetBlockContext> getLeaderBlockContextByHash(
+      final CompositeBlockKey compositeBlockKey, final boolean fetchReceipts) {
+    return getLeaderBlockContext(
+        compositeBlockKey,
+        new GetBlockByHashRequest(compositeBlockKey.getBlockHash(), fetchReceipts));
+  }
+
+  public Optional<FleetBlockContext> getLeaderBlockContextByNumber(
+      final CompositeBlockKey compositeBlockKey, final boolean fetchReceipts) {
+    return getLeaderBlockContext(
+        compositeBlockKey,
+        new GetBlockByNumberRequest(compositeBlockKey.getBlockNumber(), fetchReceipts));
+  }
+
   public void provideLeaderBlockContext(final NewHeadParams newHeadParams) {
     CompositeBlockKey key = new CompositeBlockKey(newHeadParams.getHead());
     leaderBlock.put(

src/main/java/net/consensys/fleet/follower/sync/FleetModeSynchronizer.java

@@ -115,8 +115,29 @@ private void disableWaitingSync() {
     syncDelay = 1;
   }
 
+  /**
+   * Starts the synchronization process between the follower node and the captain chain head.
+   *
+   * <p>The synchronization is divided into two main mechanisms depending on the distance from the
+   * chain head:
+   *
+   * <p>1. Far from head (when captainHead-followerHead > maxBlocksPerPersist): Blocks are fetched
+   * by block number. This approach allows retrieving a range of blocks efficiently without needing
+   * to know their hashes. If we were to fetch by block hash, we would have to walk back from the
+   * head one block at a time to discover the right hashes.
+   *
+   * <p>2. Close to head: Once synchronization is close enough to the head, it switches to fetching
+   * blocks by hash. This is more reliable and ensures we are on the correct fork.
+   *
+   * <p>Fetching by block number is safe because block hash validation is performed once we approach
+   * the head. If the remote peer (captain) is on a different fork, it will be detected during the
+   * hash-based phase, and the older blocks will be updated
+   *
+   * <p>After each range of blocks is retrieved, trielogs are applied in the same manner as in the
+   * Bonsai trie model.
+   */
   private void startSync() {
-    if (syncScheduler == null || syncScheduler.isDone()) {
+    if (syncScheduler == null || syncScheduler.isDone() || syncScheduler.isCancelled()) {
       syncScheduler =
           EXECUTOR_SERVICE.schedule(
               () -> {
@@ -142,18 +163,19 @@ private void startSync() {
                       long persistedBlockNumber = persistedBlock.getBlockHeader().getNumber();
 
                       FleetBlockContext targetBlock;
-                      if ((leaderHeader.getFinalizedBlockNumber()
+                      if ((leaderHeader.getHead().getNumber()
                               - persistedBlock.getBlockHeader().getNumber())
                           > maxBlocksPerPersist) {
                         targetBlock =
-                            getLeaderBlockContext(
+                            getLeaderBlockContextByNumber(
                                     new CompositeBlockKey(
-                                        leaderHeader.getFinalizedBlockNumber(),
-                                        leaderHeader.getFinalizedBlockHash()))
+                                        persistedBlock.getBlockHeader().getNumber()
+                                            + maxBlocksPerPersist))
                                 .orElseThrow(MissingBlockException::new);
                       } else {
                         targetBlock =
-                            getLeaderBlockContext(new CompositeBlockKey(leaderHeader.getHead()))
+                            getLeaderBlockContextByHash(
+                                    new CompositeBlockKey(leaderHeader.getHead()))
                                 .orElseThrow(MissingBlockException::new);
                       }
 
@@ -186,7 +208,7 @@ private void startSync() {
                           < targetBlock.getBlockHeader().getNumber()) {
                         LOG.debug("Rollforward {}", targetBlockHash);
                         final FleetBlockContext toRollForwardBlock =
-                            getLeaderBlockContext(
+                            getLeaderBlockContextByHash(
                                     new CompositeBlockKey(targetBlock.getBlockHeader()))
                                 .orElseThrow(MissingBlockException::new);
                         rollForward.add(toRollForwardBlock);
@@ -199,7 +221,7 @@ private void startSync() {
                           targetBlock = persistedBlock;
                         } else {
                           targetBlock =
-                              getLeaderBlockContext(
+                              getLeaderBlockContextByHash(
                                       new CompositeBlockKey(
                                           targetBlock.getBlockHeader().getNumber() - 1,
                                           targetBlock.getBlockHeader().getParentHash()))
@@ -219,11 +241,11 @@ private void startSync() {
                         LOG.debug("Paired rollforward {}", targetBlockHash);
 
                         rollForward.add(
-                            getLeaderBlockContext(
+                            getLeaderBlockContextByHash(
                                     new CompositeBlockKey(targetBlock.getBlockHeader()))
                                 .orElseThrow(MissingBlockException::new));
                         targetBlock =
-                            getLeaderBlockContext(
+                            getLeaderBlockContextByHash(
                                     new CompositeBlockKey(
                                         targetBlock.getBlockHeader().getNumber() - 1,
                                         targetBlock.getBlockHeader().getParentHash()))
@@ -390,7 +412,15 @@ private boolean isBlockchainServiceReady() {
     return pluginServiceProvider.isServiceAvailable(BlockchainService.class);
   }
 
-  private Optional<FleetBlockContext> getLeaderBlockContext(
+  private Optional<FleetBlockContext> getLeaderBlockContextByHash(
+      final CompositeBlockKey compositeBlockKey) {
+    return blockContextProvider.getLeaderBlockContextByHash(
+        compositeBlockKey,
+        (leaderHeader.getHead().getNumber() - compositeBlockKey.getBlockNumber())
+            <= headDistanceForReceiptFetch);
+  }
+
+  private Optional<FleetBlockContext> getLeaderBlockContextByNumber(
       final CompositeBlockKey compositeBlockKey) {
     return blockContextProvider.getLeaderBlockContextByNumber(
         compositeBlockKey,