Add experimental --Xsnapsync-bft-enabled which enables snap sync fo…

…r BFT chains (#7140) * Create a BFT-specific pivot block handler Signed-off-by: Matthew Whitehead <[email protected]> * Change visibility Signed-off-by: Matthew Whitehead <[email protected]> * Refactor sync-peer-count internal variable to match name, add experimental flag to enabled snap + BFT Signed-off-by: Matthew Whitehead <[email protected]> * Merge with main Signed-off-by: Matthew Whitehead <[email protected]> * Fix uppercase Signed-off-by: Matthew Whitehead <[email protected]> * Address synchronization issue with trie pruning. Create BFT-specific account range handler. Add pipeline name and logs Signed-off-by: Matthew Whitehead <[email protected]> * Remove debug log Signed-off-by: Matthew Whitehead <[email protected]> * fixing snapsync for empty state Signed-off-by: Karim Taam <[email protected]> * Don't queue up events we can't handle Signed-off-by: Matthew Whitehead <[email protected]> * Fix timing window where a validator with an empty data dir sometimes falls back to full sync if peer status isn't received quickly enough Signed-off-by: Matthew Whitehead <[email protected]> * Remove BFT-specific account request class. Not needed Signed-off-by: Matthew Whitehead <[email protected]> * Refactor some more 'fast' sync variables that are common to all pivot-based sync types Signed-off-by: Matthew Whitehead <[email protected]> * In FULL sync mode, disable bonsai-limit-trie-logs-enabled instead of failing to start Signed-off-by: Matthew Whitehead <[email protected]> * Add javadoc comments, clarify overriding bonsai-limit-trie-logs-enabled Signed-off-by: Matthew Whitehead <[email protected]> * Add BFT pivot block selector tests Signed-off-by: Matthew Whitehead <[email protected]> * Fix failure error message Signed-off-by: Matthew Whitehead <[email protected]> * Remove the unnamed Pipe constructor and update tests to set a pipe name Signed-off-by: Matthew Whitehead <[email protected]> * Revert some info logs back to debug given the feedback on noise in the logs syncing with holesky Signed-off-by: Matthew Whitehead <[email protected]> * Refactor fastSyncPivotDistance to syncPivotDistance Signed-off-by: Matthew Whitehead <[email protected]> * Incomplete refactoring Signed-off-by: Matthew Whitehead <[email protected]> * Update BFT event queueing tests Signed-off-by: Matthew Whitehead <[email protected]> * Event queue test fixes Signed-off-by: Matthew Whitehead <[email protected]> * Remove automatic setting of bonsai-limit-trie-logs-enabled to false if sync-mode = FULL (moving to another PR) Signed-off-by: Matthew Whitehead <[email protected]> --------- Signed-off-by: Matthew Whitehead <[email protected]> Signed-off-by: Karim Taam <[email protected]> Signed-off-by: Matt Whitehead <[email protected]> Co-authored-by: Karim Taam <[email protected]>
hyperledger · Jul 2, 2024 · 8ca7129 · 8ca7129
1 parent 3a73dcc
commit 8ca7129
Show file tree

Hide file tree

Showing 44 changed files with 708 additions and 127 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -33,6 +33,7 @@
 - Nodes in a permissioned chain maintain (and retry) connections to bootnodes [#7257](https://github.com/hyperledger/besu/pull/7257)
 - Promote experimental `besu storage x-trie-log` subcommand to production-ready [#7278](https://github.com/hyperledger/besu/pull/7278)
 - Enhanced BFT round-change diagnostics [#7271](https://github.com/hyperledger/besu/pull/7271)
+- `--Xsnapsync-bft-enabled` option enables experimental support for snap sync with IBFT/QBFT permissioned Bonsai-DB chains [#7140](https://github.com/hyperledger/besu/pull/7140)
 
 ### Bug fixes
 - Validation errors ignored in accounts-allowlist and empty list [#7138](https://github.com/hyperledger/besu/issues/7138)

diff --git a/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java b/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
@@ -1552,10 +1552,11 @@ private void validateOptions() {
   }
 
   private void validateConsensusSyncCompatibilityOptions() {
-    // snap and checkpoint can't be used with BFT but can for clique
-    if (genesisConfigOptionsSupplier.get().isIbftLegacy()
-        || genesisConfigOptionsSupplier.get().isIbft2()
-        || genesisConfigOptionsSupplier.get().isQbft()) {
+    // snap and checkpoint are experimental for BFT
+    if ((genesisConfigOptionsSupplier.get().isIbftLegacy()
+            || genesisConfigOptionsSupplier.get().isIbft2()
+            || genesisConfigOptionsSupplier.get().isQbft())
+        && !unstableSynchronizerOptions.isSnapSyncBftEnabled()) {
       final String errorSuffix = "can't be used with BFT networks";
       if (SyncMode.CHECKPOINT.equals(syncMode) || SyncMode.X_CHECKPOINT.equals(syncMode)) {
         throw new ParameterException(
@@ -2181,7 +2182,7 @@ private SynchronizerConfiguration buildSyncConfig() {
     return unstableSynchronizerOptions
         .toDomainObject()
         .syncMode(syncMode)
-        .fastSyncMinimumPeerCount(syncMinPeerCount)
+        .syncMinimumPeerCount(syncMinPeerCount)
         .build();
   }
 
@@ -2194,14 +2195,14 @@ private TransactionPoolConfiguration buildTransactionPoolConfiguration() {
             .saveFile((dataPath.resolve(txPoolConf.getSaveFile().getPath()).toFile()));
 
     if (genesisConfigOptionsSupplier.get().isZeroBaseFee()) {
-      logger.info(
+      logger.warn(
           "Forcing price bump for transaction replacement to 0, since we are on a zero basefee network");
       txPoolConfBuilder.priceBump(Percentage.ZERO);
     }
 
     if (miningParametersSupplier.get().getMinTransactionGasPrice().equals(Wei.ZERO)
         && !transactionPoolOptions.isPriceBumpSet(commandLine)) {
-      logger.info(
+      logger.warn(
           "Forcing price bump for transaction replacement to 0, since min-gas-price is set to 0");
       txPoolConfBuilder.priceBump(Percentage.ZERO);
     }
@@ -2810,6 +2811,7 @@ && getDataStorageConfiguration().getBonsaiLimitTrieLogsEnabled()) {
     }
 
     builder.setSnapServerEnabled(this.unstableSynchronizerOptions.isSnapsyncServerEnabled());
+    builder.setSnapSyncBftEnabled(this.unstableSynchronizerOptions.isSnapSyncBftEnabled());
 
     builder.setTxPoolImplementation(buildTransactionPoolConfiguration().getTxPoolImplementation());
     builder.setWorldStateUpdateMode(unstableEvmOptions.toDomainObject().worldUpdaterMode());

diff --git a/besu/src/main/java/org/hyperledger/besu/cli/ConfigurationOverviewBuilder.java b/besu/src/main/java/org/hyperledger/besu/cli/ConfigurationOverviewBuilder.java
@@ -56,6 +56,7 @@ public class ConfigurationOverviewBuilder {
   private long trieLogRetentionLimit = 0;
   private Integer trieLogsPruningWindowSize = null;
   private boolean isSnapServerEnabled = false;
+  private boolean isSnapSyncBftEnabled = false;
   private TransactionPoolConfiguration.Implementation txPoolImplementation;
   private EvmConfiguration.WorldUpdaterMode worldStateUpdateMode;
   private Map<String, String> environment;
@@ -233,6 +234,17 @@ public ConfigurationOverviewBuilder setSnapServerEnabled(final boolean snapServe
     return this;
   }
 
+  /**
+   * Sets snap sync BFT enabled/disabled
+   *
+   * @param snapSyncBftEnabled bool to indicate if snap sync for BFT is enabled
+   * @return the builder
+   */
+  public ConfigurationOverviewBuilder setSnapSyncBftEnabled(final boolean snapSyncBftEnabled) {
+    isSnapSyncBftEnabled = snapSyncBftEnabled;
+    return this;
+  }
+
   /**
    * Sets trie logs pruning window size
    *
@@ -357,6 +369,10 @@ public String build() {
       lines.add("Experimental Snap Sync server enabled");
     }
 
+    if (isSnapSyncBftEnabled) {
+      lines.add("Experimental Snap Sync for BFT enabled");
+    }
+
     if (isBonsaiLimitTrieLogsEnabled) {
       final StringBuilder trieLogPruningString = new StringBuilder();
       trieLogPruningString

diff --git a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
@@ -85,6 +85,8 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
 
   private static final String CHECKPOINT_POST_MERGE_FLAG = "--Xcheckpoint-post-merge-enabled";
 
+  private static final String SNAP_SYNC_BFT_ENABLED_FLAG = "--Xsnapsync-bft-enabled";
+
   /**
    * Parse block propagation range.
    *
@@ -304,6 +306,14 @@ public void parseBlockPropagationRange(final String arg) {
   private Boolean checkpointPostMergeSyncEnabled =
       SynchronizerConfiguration.DEFAULT_CHECKPOINT_POST_MERGE_ENABLED;
 
+  @CommandLine.Option(
+      names = SNAP_SYNC_BFT_ENABLED_FLAG,
+      hidden = true,
+      paramLabel = "<Boolean>",
+      arity = "0..1",
+      description = "Snap sync enabled for BFT chains (default: ${DEFAULT-VALUE})")
+  private Boolean snapsyncBftEnabled = SnapSyncConfiguration.DEFAULT_SNAP_SYNC_BFT_ENABLED;
+
   private SynchronizerOptions() {}
 
   /**
@@ -315,6 +325,15 @@ public boolean isSnapsyncServerEnabled() {
     return snapsyncServerEnabled;
   }
 
+  /**
+   * Flag to know whether the Snap sync should be enabled for a BFT chain
+   *
+   * @return true if snap sync for BFT is enabled
+   */
+  public boolean isSnapSyncBftEnabled() {
+    return snapsyncBftEnabled;
+  }
+
   /**
    * Create synchronizer options.
    *
@@ -342,7 +361,7 @@ public static SynchronizerOptions fromConfig(final SynchronizerConfiguration con
     options.downloaderParallelism = config.getDownloaderParallelism();
     options.transactionsParallelism = config.getTransactionsParallelism();
     options.computationParallelism = config.getComputationParallelism();
-    options.fastSyncPivotDistance = config.getFastSyncPivotDistance();
+    options.fastSyncPivotDistance = config.getSyncPivotDistance();
     options.fastSyncFullValidationRate = config.getFastSyncFullValidationRate();
     options.worldStateHashCountPerRequest = config.getWorldStateHashCountPerRequest();
     options.worldStateRequestParallelism = config.getWorldStateRequestParallelism();
@@ -365,6 +384,7 @@ public static SynchronizerOptions fromConfig(final SynchronizerConfiguration con
         config.getSnapSyncConfiguration().getLocalFlatStorageCountToHealPerRequest();
     options.checkpointPostMergeSyncEnabled = config.isCheckpointPostMergeEnabled();
     options.snapsyncServerEnabled = config.getSnapSyncConfiguration().isSnapServerEnabled();
+    options.snapsyncBftEnabled = config.getSnapSyncConfiguration().isSnapSyncBftEnabled();
     return options;
   }
 
@@ -380,7 +400,7 @@ public SynchronizerConfiguration.Builder toDomainObject() {
     builder.downloaderParallelism(downloaderParallelism);
     builder.transactionsParallelism(transactionsParallelism);
     builder.computationParallelism(computationParallelism);
-    builder.fastSyncPivotDistance(fastSyncPivotDistance);
+    builder.syncPivotDistance(fastSyncPivotDistance);
     builder.fastSyncFullValidationRate(fastSyncFullValidationRate);
     builder.worldStateHashCountPerRequest(worldStateHashCountPerRequest);
     builder.worldStateRequestParallelism(worldStateRequestParallelism);
@@ -397,6 +417,7 @@ public SynchronizerConfiguration.Builder toDomainObject() {
             .localFlatAccountCountToHealPerRequest(snapsyncFlatAccountHealedCountPerRequest)
             .localFlatStorageCountToHealPerRequest(snapsyncFlatStorageHealedCountPerRequest)
             .isSnapServerEnabled(snapsyncServerEnabled)
+            .isSnapSyncBftEnabled(snapsyncBftEnabled)
             .build());
     builder.checkpointPostMergeEnabled(checkpointPostMergeSyncEnabled);
 
@@ -454,7 +475,9 @@ public List<String> getCLIOptions() {
             SNAP_FLAT_STORAGE_HEALED_COUNT_PER_REQUEST_FLAG,
             OptionParser.format(snapsyncFlatStorageHealedCountPerRequest),
             SNAP_SERVER_ENABLED_FLAG,
-            OptionParser.format(snapsyncServerEnabled));
+            OptionParser.format(snapsyncServerEnabled),
+            SNAP_SYNC_BFT_ENABLED_FLAG,
+            OptionParser.format(snapsyncBftEnabled));
     return value;
   }
 }
diff --git a/besu/src/main/java/org/hyperledger/besu/controller/BesuControllerBuilder.java b/besu/src/main/java/org/hyperledger/besu/controller/BesuControllerBuilder.java
@@ -22,6 +22,7 @@
 import org.hyperledger.besu.config.GenesisConfigOptions;
 import org.hyperledger.besu.consensus.merge.MergeContext;
 import org.hyperledger.besu.consensus.merge.UnverifiedForkchoiceSupplier;
+import org.hyperledger.besu.consensus.qbft.BFTPivotSelectorFromPeers;
 import org.hyperledger.besu.consensus.qbft.pki.PkiBlockCreationConfiguration;
 import org.hyperledger.besu.cryptoservices.NodeKey;
 import org.hyperledger.besu.datatypes.Hash;
@@ -106,6 +107,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
@@ -683,7 +685,7 @@ public BesuController build() {
 
     final PivotBlockSelector pivotBlockSelector =
         createPivotSelector(
-            protocolSchedule, protocolContext, ethContext, syncState, metricsSystem);
+            protocolSchedule, protocolContext, ethContext, syncState, metricsSystem, blockchain);
 
     final Synchronizer synchronizer =
         createSynchronizer(
@@ -837,9 +839,22 @@ private PivotBlockSelector createPivotSelector(
       final ProtocolContext protocolContext,
       final EthContext ethContext,
       final SyncState syncState,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
+      final Blockchain blockchain) {
 
-    if (genesisConfigOptions.getTerminalTotalDifficulty().isPresent()) {
+    if (genesisConfigOptions.isQbft() || genesisConfigOptions.isIbft2()) {
+      LOG.info(
+          "{} is configured, creating initial sync for BFT",
+          genesisConfigOptions.getConsensusEngine().toUpperCase(Locale.ROOT));
+      return new BFTPivotSelectorFromPeers(
+          ethContext,
+          syncConfig,
+          syncState,
+          metricsSystem,
+          protocolContext,
+          nodeKey,
+          blockchain.getChainHeadHeader());
+    } else if (genesisConfigOptions.getTerminalTotalDifficulty().isPresent()) {
       LOG.info("TTD difficulty is present, creating initial sync for PoS");
 
       final MergeContext mergeContext = protocolContext.getConsensusContext(MergeContext.class);

diff --git a/besu/src/test/java/org/hyperledger/besu/RunnerTest.java b/besu/src/test/java/org/hyperledger/besu/RunnerTest.java
@@ -225,8 +225,8 @@ private void syncFromGenesis(final SyncMode mode, final GenesisConfigFile genesi
       final SynchronizerConfiguration syncConfigBehind =
           SynchronizerConfiguration.builder()
               .syncMode(mode)
-              .fastSyncPivotDistance(5)
-              .fastSyncMinimumPeerCount(1)
+              .syncPivotDistance(5)
+              .syncMinimumPeerCount(1)
               .build();
       final Path dataDirBehind = Files.createTempDirectory(temp, "db-behind");
 

diff --git a/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java b/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java
@@ -1167,7 +1167,7 @@ public void checkValidDefaultFastSyncMinPeers() {
 
     final SynchronizerConfiguration syncConfig = syncConfigurationCaptor.getValue();
     assertThat(syncConfig.getSyncMode()).isEqualTo(SyncMode.FAST);
-    assertThat(syncConfig.getFastSyncMinimumPeerCount()).isEqualTo(5);
+    assertThat(syncConfig.getSyncMinimumPeerCount()).isEqualTo(5);
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
   }
@@ -1179,7 +1179,7 @@ public void parsesValidFastSyncMinPeersOption() {
 
     final SynchronizerConfiguration syncConfig = syncConfigurationCaptor.getValue();
     assertThat(syncConfig.getSyncMode()).isEqualTo(SyncMode.FAST);
-    assertThat(syncConfig.getFastSyncMinimumPeerCount()).isEqualTo(11);
+    assertThat(syncConfig.getSyncMinimumPeerCount()).isEqualTo(11);
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
   }
@@ -1191,7 +1191,7 @@ public void parsesValidSnapSyncMinPeersOption() {
 
     final SynchronizerConfiguration syncConfig = syncConfigurationCaptor.getValue();
     assertThat(syncConfig.getSyncMode()).isEqualTo(SyncMode.SNAP);
-    assertThat(syncConfig.getFastSyncMinimumPeerCount()).isEqualTo(11);
+    assertThat(syncConfig.getSyncMinimumPeerCount()).isEqualTo(11);
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
   }
@@ -1203,7 +1203,7 @@ public void parsesValidSyncMinPeersOption() {
 
     final SynchronizerConfiguration syncConfig = syncConfigurationCaptor.getValue();
     assertThat(syncConfig.getSyncMode()).isEqualTo(SyncMode.FAST);
-    assertThat(syncConfig.getFastSyncMinimumPeerCount()).isEqualTo(11);
+    assertThat(syncConfig.getSyncMinimumPeerCount()).isEqualTo(11);
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
   }

diff --git a/besu/src/test/java/org/hyperledger/besu/cli/CascadingDefaultProviderTest.java b/besu/src/test/java/org/hyperledger/besu/cli/CascadingDefaultProviderTest.java
@@ -138,7 +138,7 @@ public void overrideDefaultValuesIfKeyIsPresentInConfigFile(final @TempDir File
     verify(mockControllerBuilder).synchronizerConfiguration(syncConfigurationCaptor.capture());
 
     assertThat(syncConfigurationCaptor.getValue().getSyncMode()).isEqualTo(SyncMode.FAST);
-    assertThat(syncConfigurationCaptor.getValue().getFastSyncMinimumPeerCount()).isEqualTo(13);
+    assertThat(syncConfigurationCaptor.getValue().getSyncMinimumPeerCount()).isEqualTo(13);
 
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
@@ -182,7 +182,7 @@ public void noOverrideDefaultValuesIfKeyIsNotPresentInConfigFile() {
 
     final SynchronizerConfiguration syncConfig = syncConfigurationCaptor.getValue();
     assertThat(syncConfig.getSyncMode()).isEqualTo(SyncMode.SNAP);
-    assertThat(syncConfig.getFastSyncMinimumPeerCount()).isEqualTo(5);
+    assertThat(syncConfig.getSyncMinimumPeerCount()).isEqualTo(5);
 
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();

diff --git a/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java b/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java
@@ -38,9 +38,9 @@ protected SynchronizerConfiguration.Builder createDefaultDomainObject() {
   @Override
   protected SynchronizerConfiguration.Builder createCustomizedDomainObject() {
     return SynchronizerConfiguration.builder()
-        .fastSyncPivotDistance(SynchronizerConfiguration.DEFAULT_PIVOT_DISTANCE_FROM_HEAD + 10)
+        .syncPivotDistance(SynchronizerConfiguration.DEFAULT_PIVOT_DISTANCE_FROM_HEAD + 10)
         .fastSyncFullValidationRate(SynchronizerConfiguration.DEFAULT_FULL_VALIDATION_RATE / 2)
-        .fastSyncMinimumPeerCount(SynchronizerConfiguration.DEFAULT_FAST_SYNC_MINIMUM_PEERS + 2)
+        .syncMinimumPeerCount(SynchronizerConfiguration.DEFAULT_SYNC_MINIMUM_PEERS + 2)
         .worldStateHashCountPerRequest(
             SynchronizerConfiguration.DEFAULT_WORLD_STATE_HASH_COUNT_PER_REQUEST + 2)
         .worldStateRequestParallelism(
@@ -94,7 +94,7 @@ protected SynchronizerOptions getOptionsFromBesuCommand(final TestBesuCommand be
 
   @Override
   protected List<String> getFieldsToIgnore() {
-    return Arrays.asList("fastSyncMinimumPeerCount");
+    return Arrays.asList("syncMinimumPeerCount");
   }
 
   @Override

diff --git a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BftEventQueue.java b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BftEventQueue.java
@@ -15,10 +15,12 @@
 package org.hyperledger.besu.consensus.common.bft;
 
 import org.hyperledger.besu.consensus.common.bft.events.BftEvent;
+import org.hyperledger.besu.consensus.common.bft.events.BftEvents;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nullable;
 
 import org.slf4j.Logger;
@@ -30,6 +32,7 @@ public class BftEventQueue {
 
   private static final Logger LOG = LoggerFactory.getLogger(BftEventQueue.class);
   private final int messageQueueLimit;
+  private final AtomicBoolean started = new AtomicBoolean(false);
 
   /**
    * Instantiates a new Bft event queue.
@@ -40,16 +43,30 @@ public BftEventQueue(final int messageQueueLimit) {
     this.messageQueueLimit = messageQueueLimit;
   }
 
+  /** Start the event queue. Until it has been started no events will be queued for processing. */
+  public void start() {
+    started.set(true);
+  }
+
+  private boolean isStarted() {
+    return started.get();
+  }
+
   /**
-   * Put an Bft event onto the queue
+   * Put an Bft event onto the queue. Note: the event queue must be started before an event will be
+   * queued for processing. Events received before the queue is started will be discarded.
    *
    * @param event Provided bft event
    */
   public void add(final BftEvent event) {
-    if (queue.size() > messageQueueLimit) {
-      LOG.warn("Queue size exceeded trying to add new bft event {}", event);
-    } else {
-      queue.add(event);
+
+    // Don't queue events other than block timer expiry, until we know we can process them
+    if (isStarted() || event.getType() == BftEvents.Type.BLOCK_TIMER_EXPIRY) {
+      if (queue.size() > messageQueueLimit) {
+        LOG.warn("Queue size exceeded trying to add new bft event {}", event);
+      } else {
+        queue.add(event);
+      }
     }
   }
 

diff --git a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BftProcessor.java b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BftProcessor.java
@@ -61,6 +61,9 @@ public void awaitStop() throws InterruptedException {
   @Override
   public void run() {
     try {
+      // Start the event queue. Until it is started it won't accept new events from peers
+      incomingQueue.start();
+
       while (!shutdown) {
         nextEvent().ifPresent(eventMultiplexer::handleBftEvent);
       }