Stratum mining errata (hyperledger#3554)

* couple bits of additional plumbing to ensure transitionCoordinator correctly routes stratum ethash observers

Signed-off-by: garyschulte <garyschulte@gmail.com>

besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -23,6 +23,7 @@
 
 import org.hyperledger.besu.cli.config.EthNetworkConfig;
 import org.hyperledger.besu.cli.config.NetworkName;
+import org.hyperledger.besu.consensus.merge.blockcreation.TransitionCoordinator;
 import org.hyperledger.besu.controller.BesuController;
 import org.hyperledger.besu.crypto.NodeKey;
 import org.hyperledger.besu.ethereum.ProtocolContext;
@@ -537,17 +538,24 @@ public Runner build() {
 
  final MiningParameters miningParameters = besuController.getMiningParameters();
  Optional<StratumServer> stratumServer = Optional.empty();
+
  if (miningParameters.isStratumMiningEnabled()) {
+ var powMiningCoordinator = miningCoordinator;
+ if (miningCoordinator instanceof TransitionCoordinator) {
+ LOG.debug("fetching powMiningCoordinator from TransitionCoordinator");
+ powMiningCoordinator = ((TransitionCoordinator) miningCoordinator).getPreMergeObject();
+ }
  stratumServer =
  Optional.of(
  new StratumServer(
  vertx,
- miningCoordinator,
+ powMiningCoordinator,
  miningParameters.getStratumPort(),
  miningParameters.getStratumNetworkInterface(),
  miningParameters.getStratumExtranonce(),
  metricsSystem));
  miningCoordinator.addEthHashObserver(stratumServer.get());
+ LOG.debug("added ethash observer: {}", stratumServer.get());
  }
 
  sanitizePeers(network, staticNodes)

besu/src/test/java/org/hyperledger/besu/RunnerBuilderTest.java

@@ -17,7 +17,11 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider.createInMemoryBlockchain;
 import static org.hyperledger.besu.ethereum.storage.keyvalue.KeyValueSegmentIdentifier.BLOCKCHAIN;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.cli.config.EthNetworkConfig;
@@ -27,6 +31,7 @@
 import org.hyperledger.besu.consensus.common.bft.protocol.BftProtocolManager;
 import org.hyperledger.besu.consensus.ibft.protocol.IbftSubProtocol;
 import org.hyperledger.besu.consensus.merge.blockcreation.MergeMiningCoordinator;
+import org.hyperledger.besu.consensus.merge.blockcreation.TransitionCoordinator;
 import org.hyperledger.besu.controller.BesuController;
 import org.hyperledger.besu.crypto.KeyPairSecurityModule;
 import org.hyperledger.besu.crypto.NodeKey;
@@ -36,6 +41,7 @@
 import org.hyperledger.besu.ethereum.api.jsonrpc.JsonRpcConfiguration;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.WebSocketConfiguration;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
+import org.hyperledger.besu.ethereum.blockcreation.PoWMiningCoordinator;
 import org.hyperledger.besu.ethereum.chain.DefaultBlockchain;
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.Block;
@@ -333,4 +339,49 @@ public void noEngineApiNoServiceForMethods() {
  assertThat(runner.getEngineJsonRpcPort()).isEmpty();
  assertThat(runner.getEngineWebsocketPort()).isEmpty();
  }
+
+ @Test
+ public void assertTransitionStratumConfiguration() {
+ JsonRpcConfiguration jrpc = JsonRpcConfiguration.createDefault();
+ jrpc.setEnabled(true);
+ JsonRpcConfiguration engine = JsonRpcConfiguration.createEngineDefault();
+ engine.setEnabled(true);
+ EthNetworkConfig mockMainnet = mock(EthNetworkConfig.class);
+ when(mockMainnet.getNetworkId()).thenReturn(BigInteger.ONE);
+ MergeConfigOptions.setMergeEnabled(true);
+ MiningParameters mockMiningParams = besuController.getMiningParameters();
+ when(mockMiningParams.isStratumMiningEnabled()).thenReturn(true);
+ TransitionCoordinator mockTransitionCoordinator =
+ spy(
+ new TransitionCoordinator(
+ mock(PoWMiningCoordinator.class), mock(MergeMiningCoordinator.class)));
+ when(besuController.getMiningCoordinator()).thenReturn(mockTransitionCoordinator);
+
+ new RunnerBuilder()
+ .discovery(true)
+ .p2pListenInterface("0.0.0.0")
+ .p2pListenPort(30303)
+ .p2pAdvertisedHost("127.0.0.1")
+ .p2pEnabled(true)
+ .natMethod(NatMethod.NONE)
+ .besuController(besuController)
+ .ethNetworkConfig(mockMainnet)
+ .metricsSystem(mock(ObservableMetricsSystem.class))
+ .permissioningService(mock(PermissioningServiceImpl.class))
+ .jsonRpcConfiguration(jrpc)
+ .engineJsonRpcConfiguration(engine)
+ .graphQLConfiguration(mock(GraphQLConfiguration.class))
+ .webSocketConfiguration(mock(WebSocketConfiguration.class))
+ .metricsConfiguration(mock(MetricsConfiguration.class))
+ .vertx(Vertx.vertx())
+ .dataDir(dataDir.getRoot().toPath())
+ .storageProvider(mock(KeyValueStorageProvider.class))
+ .forkIdSupplier(() -> Collections.singletonList(Bytes.EMPTY))
+ .rpcEndpointService(new RpcEndpointServiceImpl())
+ .besuPluginContext(mock(BesuPluginContextImpl.class))
+ .build();
+
+ verify(mockTransitionCoordinator, times(1)).getPreMergeObject();
+ verify(mockTransitionCoordinator, times(1)).addEthHashObserver(any());
+ }
 }

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/blockcreation/TransitionCoordinator.java

@@ -20,6 +20,8 @@
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.BlockValidator.Result;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
+import org.hyperledger.besu.ethereum.blockcreation.PoWMiningCoordinator;
+import org.hyperledger.besu.ethereum.chain.PoWObserver;
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
@@ -107,6 +109,13 @@ public Optional<Block> createBlock(final BlockHeader parentHeader, final long ti
  (MiningCoordinator coordinator) -> coordinator.createBlock(parentHeader, timestamp));
  }
 
+ @Override
+ public void addEthHashObserver(final PoWObserver observer) {
+ if (this.miningCoordinator instanceof PoWMiningCoordinator) {
+ miningCoordinator.addEthHashObserver(observer);
+ }
+ }
+
  @Override
  public void changeTargetGasLimit(final Long targetGasLimit) {
  miningCoordinator.changeTargetGasLimit(targetGasLimit);

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/PoWSolver.java

@@ -109,8 +109,11 @@ public PoWSolution solveFor(final PoWSolverJob job)
  currentJobs.put(
  job.getInputs().getPrePowHash(), job, System.currentTimeMillis() + powJobTimeToLive);
  if (stratumMiningEnabled) {
+ LOG.debug(
+ "solving with stratum miner for {} observers", ethHashObservers.getSubscriberCount());
  ethHashObservers.forEach(observer -> observer.newJob(job.inputs));
  } else {
+ LOG.debug("solving with cpu miner");
  findValidNonce();
  }
  return job.getSolution();

ethereum/stratum/src/main/java/org/hyperledger/besu/ethereum/stratum/GetWorkProtocol.java

@@ -14,6 +14,8 @@
  */
 package org.hyperledger.besu.ethereum.stratum;
 
+import static org.hyperledger.besu.util.Slf4jLambdaHelper.debugLambda;
+
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.Quantity;
@@ -156,6 +158,7 @@ public void handle(final StratumConnection conn, final String message) {
 
  @Override
  public void setCurrentWorkTask(final PoWSolverInputs input) {
+ debugLambda(LOG, "setting current stratum work task {}", input::toString);
  currentInput = input;
  final byte[] dagSeed =
  DirectAcyclicGraphSeed.dagSeed(currentInput.getBlockNumber(), epochCalculator);

ethereum/stratum/src/main/java/org/hyperledger/besu/ethereum/stratum/StratumServer.java

@@ -157,6 +157,7 @@ public void newJob(final PoWSolverInputs poWSolverInputs) {
  logger.debug("Discarding {} as stratum server is not started", poWSolverInputs);
  return;
  }
+ logger.debug("stratum newJob with inputs: {}", poWSolverInputs);
  for (StratumProtocol protocol : protocols) {
  protocol.setCurrentWorkTask(poWSolverInputs);
  }