QBFT migration: Create schedulable mining coordinator

#2999

Signed-off-by: Simon Dudley <simon.dudley@consensys.net>

besu/src/main/java/org/hyperledger/besu/controller/ConsensusScheduleBesuControllerBuilder.java

@@ -21,6 +21,9 @@
 import org.hyperledger.besu.config.GenesisConfigFile;
 import org.hyperledger.besu.consensus.common.CombinedProtocolScheduleFactory;
 import org.hyperledger.besu.consensus.common.ForkSpec;
+import org.hyperledger.besu.consensus.common.bft.BftForksSchedule;
+import org.hyperledger.besu.consensus.common.bft.blockcreation.BftMiningCoordinator;
+import org.hyperledger.besu.consensus.common.bft.blockcreation.SchedulableBftMiningCoordinator;
 import org.hyperledger.besu.consensus.qbft.pki.PkiBlockCreationConfiguration;
 import org.hyperledger.besu.crypto.NodeKey;
 import org.hyperledger.besu.datatypes.Hash;
@@ -104,6 +107,7 @@ protected ConsensusScheduleBesuControllerBuilder(
   @Override
   protected void prepForBuild() {
     besuControllerBuilderSchedule.values().forEach(BesuControllerBuilder::prepForBuild);
+    super.prepForBuild();
   }
 
   @Override
@@ -114,15 +118,33 @@ protected MiningCoordinator createMiningCoordinator(
       final MiningParameters miningParameters,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
-    return besuControllerBuilderSchedule
-        .get(0L)
-        .createMiningCoordinator(
-            protocolSchedule,
-            protocolContext,
-            transactionPool,
-            miningParameters,
-            syncState,
-            ethProtocolManager);
+
+    final List<ForkSpec<BftMiningCoordinator>> miningCoordinatorForkSpecs =
+        besuControllerBuilderSchedule.entrySet().stream()
+            // TODO SLD need to make this work for NoopMiningCoordinator (for IBFT1 -> QBFT
+            // migration)
+            // onBlockAdded is the problematic method
+            .map(
+                e ->
+                    new ForkSpec<>(
+                        e.getKey(),
+                        (BftMiningCoordinator)
+                            e.getValue()
+                                .createMiningCoordinator(
+                                    protocolSchedule,
+                                    protocolContext,
+                                    transactionPool,
+                                    miningParameters,
+                                    syncState,
+                                    ethProtocolManager)))
+            .collect(Collectors.toList());
+    final BftForksSchedule<BftMiningCoordinator> miningCoordinatorSchedule =
+        new BftForksSchedule<>(
+            miningCoordinatorForkSpecs.get(0),
+            miningCoordinatorForkSpecs.subList(1, miningCoordinatorForkSpecs.size()));
+
+    return new SchedulableBftMiningCoordinator(
+        miningCoordinatorSchedule, protocolContext.getBlockchain());
   }
 
   @Override
@@ -169,7 +191,7 @@ protected SubProtocolConfiguration createSubProtocolConfiguration(
 
   @Override
   protected void validateContext(final ProtocolContext context) {
-    besuControllerBuilderSchedule.get(0L).validateContext(context);
+    besuControllerBuilderSchedule.values().forEach(builder -> builder.validateContext(context));
   }
 
   @Override

besu/src/test/java/org/hyperledger/besu/controller/ConsensusScheduleBesuControllerBuilderTest.java

@@ -17,12 +17,24 @@
  */
 package org.hyperledger.besu.controller;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.config.GenesisConfigFile;
 import org.hyperledger.besu.config.StubGenesisConfigOptions;
 import org.hyperledger.besu.consensus.common.ForkSpec;
+import org.hyperledger.besu.consensus.common.bft.blockcreation.BftMiningCoordinator;
+import org.hyperledger.besu.consensus.common.bft.blockcreation.SchedulableBftMiningCoordinator;
+import org.hyperledger.besu.ethereum.ProtocolContext;
+import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
+import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
+import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 
 import java.math.BigInteger;
@@ -34,6 +46,7 @@
 import java.util.TreeSet;
 import java.util.function.BiFunction;
 
+import org.assertj.core.api.SoftAssertions;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
@@ -52,6 +65,8 @@ public class ConsensusScheduleBesuControllerBuilderTest {
   private @Mock ProtocolSchedule protocolSchedule1;
   private @Mock ProtocolSchedule protocolSchedule2;
   private @Mock ProtocolSchedule protocolSchedule3;
+  private @Mock BftMiningCoordinator bftMiningCoordinator1;
+  private @Mock BftMiningCoordinator bftMiningCoordinator2;
 
   @Test
   public void mustProvideNonNullConsensusScheduleWhenInstantiatingNew() {
@@ -98,4 +113,64 @@ public void mustCreateCombinedProtocolScheduleUsingProtocolSchedulesOrderedByBlo
     Mockito.verify(combinedProtocolScheduleFactory)
         .apply(expectedProtocolSchedulesSpecs, Optional.of(BigInteger.TEN));
   }
+
+  @Test
+  public void createsScheduableMiningCoordinator() {
+    final Map<Long, BesuControllerBuilder> consensusSchedule =
+        Map.of(0L, besuControllerBuilder1, 5L, besuControllerBuilder2);
+
+    when(besuControllerBuilder1.createMiningCoordinator(any(), any(), any(), any(), any(), any()))
+        .thenReturn(bftMiningCoordinator1);
+    when(besuControllerBuilder2.createMiningCoordinator(any(), any(), any(), any(), any(), any()))
+        .thenReturn(bftMiningCoordinator2);
+    final ProtocolContext mockProtocolContext = mock(ProtocolContext.class);
+    when(mockProtocolContext.getBlockchain()).thenReturn(mock(MutableBlockchain.class));
+
+    final ConsensusScheduleBesuControllerBuilder builder =
+        new ConsensusScheduleBesuControllerBuilder(consensusSchedule);
+    final MiningCoordinator miningCoordinator =
+        builder.createMiningCoordinator(
+            protocolSchedule1,
+            mockProtocolContext,
+            mock(TransactionPool.class),
+            mock(MiningParameters.class),
+            mock(SyncState.class),
+            mock(EthProtocolManager.class));
+
+    assertThat(miningCoordinator).isInstanceOf(SchedulableBftMiningCoordinator.class);
+    final SchedulableBftMiningCoordinator schedulableBftMiningCoordinator =
+        (SchedulableBftMiningCoordinator) miningCoordinator;
+
+    SoftAssertions.assertSoftly(
+        (softly) -> {
+          softly
+              .assertThat(
+                  schedulableBftMiningCoordinator
+                      .getMiningCoordinatorSchedule()
+                      .getFork(0L)
+                      .getValue())
+              .isSameAs(bftMiningCoordinator1);
+          softly
+              .assertThat(
+                  schedulableBftMiningCoordinator
+                      .getMiningCoordinatorSchedule()
+                      .getFork(4L)
+                      .getValue())
+              .isSameAs(bftMiningCoordinator1);
+          softly
+              .assertThat(
+                  schedulableBftMiningCoordinator
+                      .getMiningCoordinatorSchedule()
+                      .getFork(5L)
+                      .getValue())
+              .isSameAs(bftMiningCoordinator2);
+          softly
+              .assertThat(
+                  schedulableBftMiningCoordinator
+                      .getMiningCoordinatorSchedule()
+                      .getFork(6L)
+                      .getValue())
+              .isSameAs(bftMiningCoordinator2);
+        });
+  }
 }

consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BftForksSchedule.java

@@ -31,7 +31,8 @@
 
 import com.google.common.annotations.VisibleForTesting;
 
-public class BftForksSchedule<C extends BftConfigOptions> {
+// TODO SLD if using this with <MiningCoordinator>, should I rename this to ForksSchedule?
+public class BftForksSchedule<C> {
 
   private final NavigableSet<ForkSpec<C>> forks =
       new TreeSet<>(

consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/blockcreation/BftMiningCoordinator.java

@@ -33,6 +33,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.logging.log4j.Logger;
@@ -78,7 +79,7 @@ public BftMiningCoordinator(
   public void start() {
     if (state.compareAndSet(State.IDLE, State.RUNNING)) {
       bftExecutors.start();
-      blockAddedObserverId = blockchain.observeBlockAdded(this);
+      blockAddedObserverId = blockchain.observeBlockAdded(this); // TODO SLD duplicate block event
       eventHandler.start();
       bftExecutors.executeBftProcessor(bftProcessor);
     }
@@ -90,13 +91,19 @@ public void stop() {
       blockchain.removeObserver(blockAddedObserverId);
       bftProcessor.stop();
       // Make sure the processor has stopped before shutting down the executors
-      try {
-        bftProcessor.awaitStop();
-      } catch (final InterruptedException e) {
-        LOG.debug("Interrupted while waiting for IbftProcessor to stop.", e);
-        Thread.currentThread().interrupt();
-      }
-      bftExecutors.stop();
+      // TODO this may not complete before the next miningCoordinator is started
+      // workaround until this is complete: https://github.com/hyperledger/besu/issues/3003
+      Executors.newSingleThreadExecutor()
+          .submit(
+              () -> {
+                try {
+                  bftProcessor.awaitStop();
+                } catch (final InterruptedException e) {
+                  LOG.debug("Interrupted while waiting for IbftProcessor to stop.", e);
+                  Thread.currentThread().interrupt();
+                }
+                bftExecutors.stop();
+              });
     }
   }
 

consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/blockcreation/SchedulableBftMiningCoordinator.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * 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 org.hyperledger.besu.consensus.common.bft.blockcreation;
+
+import static org.apache.logging.log4j.LogManager.getLogger;
+
+import org.hyperledger.besu.consensus.common.bft.BftForksSchedule;
+import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.Wei;
+import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
+import org.hyperledger.besu.ethereum.chain.BlockAddedEvent;
+import org.hyperledger.besu.ethereum.chain.BlockAddedObserver;
+import org.hyperledger.besu.ethereum.chain.Blockchain;
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.core.Transaction;
+
+import java.util.List;
+import java.util.Optional;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.logging.log4j.Logger;
+import org.apache.tuweni.bytes.Bytes;
+
+public class SchedulableBftMiningCoordinator implements MiningCoordinator, BlockAddedObserver {
+
+  private static final Logger LOG = getLogger();
+
+  private final BftForksSchedule<BftMiningCoordinator> miningCoordinatorSchedule;
+  private final Blockchain blockchain;
+  private BftMiningCoordinator activeMiningCoordinator;
+  private long blockAddedObserverId;
+
+  public SchedulableBftMiningCoordinator(
+      final BftForksSchedule<BftMiningCoordinator> miningCoordinatorSchedule,
+      final Blockchain blockchain) {
+    this.miningCoordinatorSchedule = miningCoordinatorSchedule;
+    this.blockchain = blockchain;
+    this.activeMiningCoordinator =
+        this.miningCoordinatorSchedule.getFork(blockchain.getChainHeadBlockNumber()).getValue();
+  }
+
+  @Override
+  public void start() {
+    blockAddedObserverId = blockchain.observeBlockAdded(this);
+    activeMiningCoordinator.start();
+  }
+
+  @Override
+  public void stop() {
+    blockchain.removeObserver(blockAddedObserverId);
+    activeMiningCoordinator.stop();
+  }
+
+  @Override
+  public void awaitStop() throws InterruptedException {
+    activeMiningCoordinator.awaitStop();
+  }
+
+  @Override
+  public boolean enable() {
+    return activeMiningCoordinator.enable();
+  }
+
+  @Override
+  public boolean disable() {
+    return activeMiningCoordinator.disable();
+  }
+
+  @Override
+  public boolean isMining() {
+    return activeMiningCoordinator.isMining();
+  }
+
+  @Override
+  public Wei getMinTransactionGasPrice() {
+    return activeMiningCoordinator.getMinTransactionGasPrice();
+  }
+
+  @Override
+  public void setExtraData(final Bytes extraData) {
+    activeMiningCoordinator.setExtraData(extraData);
+  }
+
+  @Override
+  public Optional<Address> getCoinbase() {
+    return activeMiningCoordinator.getCoinbase();
+  }
+
+  @Override
+  public Optional<Block> createBlock(
+      final BlockHeader parentHeader,
+      final List<Transaction> transactions,
+      final List<BlockHeader> ommers) {
+    return activeMiningCoordinator.createBlock(parentHeader, transactions, ommers);
+  }
+
+  @Override
+  public void changeTargetGasLimit(final Long targetGasLimit) {
+    activeMiningCoordinator.changeTargetGasLimit(targetGasLimit);
+  }
+
+  @Override
+  public void onBlockAdded(final BlockAddedEvent event) {
+    final long currentBlock = event.getBlock().getHeader().getNumber();
+    final BftMiningCoordinator nextMiningCoordinator =
+        miningCoordinatorSchedule.getFork(currentBlock + 1).getValue();
+    if (!activeMiningCoordinator.equals(
+        nextMiningCoordinator)) { // TODO SLD equals() or object ref equality?
+      LOG.debug(
+          "Switching mining coordinator after block {} from {} to {}",
+          currentBlock,
+          activeMiningCoordinator.getClass().getName(),
+          nextMiningCoordinator.getClass().getTypeName());
+      activeMiningCoordinator.stop();
+      nextMiningCoordinator.start();
+      activeMiningCoordinator = nextMiningCoordinator;
+    }
+    activeMiningCoordinator.onBlockAdded(event);
+  }
+
+  @VisibleForTesting
+  public BftForksSchedule<BftMiningCoordinator> getMiningCoordinatorSchedule() {
+    return this.miningCoordinatorSchedule;
+  }
+}