Time limited block creation (#6044)

Signed-off-by: Fabio Di Fabio <fabio.difabio@consensys.net>
Co-authored-by: Sally MacFarlane <macfarla.github@gmail.com>

CHANGELOG.md

@@ -18,6 +18,7 @@
 - Upgrade EVM Reference Tests to v13 (Cancun) [#6114](https://github.com/hyperledger/besu/pull/6114)
 - Add `yParity` to GraphQL and JSON-RPC for relevant querise. [6119](https://github.com/hyperledger/besu/pull/6119)
 - Force tx replacement price bump to zero when zero base fee market is configured or `--min-gas-price` is set to 0. This allows for easier tx replacement in networks where there is not gas price. [#6079](https://github.com/hyperledger/besu/pull/6079)
+- Introduce the possibility to limit the time spent selecting pending transactions during block creation, using the new experimental option `Xblock-txs-selection-max-time` on PoS and PoW networks (by default set to 5000ms) or `Xpoa-block-txs-selection-max-time` on PoA networks (by default 75% of the min block time) [#6044](https://github.com/hyperledger/besu/pull/6044)
 
 ### Bug fixes
 - Upgrade netty to address CVE-2023-44487, CVE-2023-34462 [#6100](https://github.com/hyperledger/besu/pull/6100)

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -122,6 +122,7 @@ import org.hyperledger.besu.ethereum.api.tls.FileBasedPasswordProvider;
 import org.hyperledger.besu.ethereum.api.tls.TlsClientAuthConfiguration;
 import org.hyperledger.besu.ethereum.api.tls.TlsConfiguration;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
+import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.eth.sync.SyncMode;
@@ -1806,8 +1807,7 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
   }
 
   private void validateMiningParams() {
-    miningOptions.validate(
-        commandLine, logger, isMergeEnabled(), getActualGenesisConfigOptions().isEthHash());
+    miningOptions.validate(commandLine, getActualGenesisConfigOptions(), isMergeEnabled(), logger);
   }
 
   /**
@@ -2830,11 +2830,36 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
 
   private MiningParameters getMiningParameters() {
     if (miningParameters == null) {
-      miningParameters = miningOptions.toDomainObject();
+      final var miningParametersBuilder =
+          ImmutableMiningParameters.builder().from(miningOptions.toDomainObject());
+      final var actualGenesisOptions = getActualGenesisConfigOptions();
+      if (actualGenesisOptions.isPoa()) {
+        miningParametersBuilder.unstable(
+            ImmutableMiningParameters.Unstable.builder()
+                .minBlockTime(getMinBlockTime(actualGenesisOptions))
+                .build());
+      }
+      miningParameters = miningParametersBuilder.build();
     }
     return miningParameters;
   }
 
+  private int getMinBlockTime(final GenesisConfigOptions genesisConfigOptions) {
+    if (genesisConfigOptions.isClique()) {
+      return genesisConfigOptions.getCliqueConfigOptions().getBlockPeriodSeconds();
+    }
+
+    if (genesisConfigOptions.isIbft2()) {
+      return genesisConfigOptions.getBftConfigOptions().getBlockPeriodSeconds();
+    }
+
+    if (genesisConfigOptions.isQbft()) {
+      return genesisConfigOptions.getQbftConfigOptions().getBlockPeriodSeconds();
+    }
+
+    throw new IllegalArgumentException("Should only be called for a PoA network");
+  }
+
   private boolean isPruningEnabled() {
     return pruningEnabled;
   }

besu/src/main/java/org/hyperledger/besu/cli/options/MiningOptions.java

@@ -15,23 +15,29 @@
 package org.hyperledger.besu.cli.options;
 
 import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.MutableInitValues.DEFAULT_EXTRA_DATA;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.MutableInitValues.DEFAULT_MIN_BLOCK_OCCUPANCY_RATIO;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.MutableInitValues.DEFAULT_MIN_PRIORITY_FEE_PER_GAS;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.MutableInitValues.DEFAULT_MIN_TRANSACTION_GAS_PRICE;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_MAX_OMMERS_DEPTH;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POS_BLOCK_CREATION_MAX_TIME;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POS_BLOCK_CREATION_REPETITION_MIN_DURATION;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POW_JOB_TTL;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_REMOTE_SEALERS_LIMIT;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_REMOTE_SEALERS_TTL;
 
+import org.hyperledger.besu.cli.converter.PercentageConverter;
 import org.hyperledger.besu.cli.util.CommandLineUtils;
+import org.hyperledger.besu.config.GenesisConfigOptions;
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitValues;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.util.number.Percentage;
 
 import java.util.List;
 
@@ -162,6 +168,25 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
                 + " then it waits before next repetition. Must be positive and ≤ 2000 (default: ${DEFAULT-VALUE} milliseconds)")
     private Long posBlockCreationRepetitionMinDuration =
         DEFAULT_POS_BLOCK_CREATION_REPETITION_MIN_DURATION;
+
+    @CommandLine.Option(
+        hidden = true,
+        names = {"--Xblock-txs-selection-max-time"},
+        description =
+            "Specifies the maximum time, in milliseconds, that could be spent selecting transactions to be included in the block."
+                + " Not compatible with PoA networks, see Xpoa-block-txs-selection-max-time."
+                + " Must be positive and ≤ (default: ${DEFAULT-VALUE})")
+    private Long nonPoaBlockTxsSelectionMaxTime = DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
+
+    @CommandLine.Option(
+        hidden = true,
+        names = {"--Xpoa-block-txs-selection-max-time"},
+        converter = PercentageConverter.class,
+        description =
+            "Specifies the maximum time that could be spent selecting transactions to be included in the block, as a percentage of the fixed block time of the PoA network."
+                + " To be only used on PoA networks, for other networks see Xblock-txs-selection-max-time."
+                + " (default: ${DEFAULT-VALUE})")
+    private Percentage poaBlockTxsSelectionMaxTime = DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME;
   }
 
   private MiningOptions() {}
@@ -180,15 +205,15 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
    * options are valid for the selected implementation.
    *
    * @param commandLine the full commandLine to check all the options specified by the user
-   * @param logger the logger
+   * @param genesisConfigOptions is EthHash?
    * @param isMergeEnabled is the Merge enabled?
-   * @param isEthHash is EthHash?
+   * @param logger the logger
    */
   public void validate(
       final CommandLine commandLine,
-      final Logger logger,
+      final GenesisConfigOptions genesisConfigOptions,
       final boolean isMergeEnabled,
-      final boolean isEthHash) {
+      final Logger logger) {
     if (Boolean.TRUE.equals(isMiningEnabled) && coinbase == null) {
       throw new ParameterException(
           commandLine,
@@ -203,7 +228,7 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
     }
 
     // Check that block producer options work
-    if (!isMergeEnabled && isEthHash) {
+    if (!isMergeEnabled && genesisConfigOptions.isEthHash()) {
       CommandLineUtils.checkOptionDependencies(
           logger,
           commandLine,
@@ -231,7 +256,9 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
     if (unstableOptions.posBlockCreationMaxTime <= 0
         || unstableOptions.posBlockCreationMaxTime > DEFAULT_POS_BLOCK_CREATION_MAX_TIME) {
       throw new ParameterException(
-          commandLine, "--Xpos-block-creation-max-time must be positive and ≤ 12000");
+          commandLine,
+          "--Xpos-block-creation-max-time must be positive and ≤ "
+              + DEFAULT_POS_BLOCK_CREATION_MAX_TIME);
     }
 
     if (unstableOptions.posBlockCreationRepetitionMinDuration <= 0
@@ -239,6 +266,31 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
       throw new ParameterException(
           commandLine, "--Xpos-block-creation-repetition-min-duration must be positive and ≤ 2000");
     }
+
+    if (genesisConfigOptions.isPoa()) {
+      CommandLineUtils.failIfOptionDoesntMeetRequirement(
+          commandLine,
+          "--Xblock-txs-selection-max-time can't be used with PoA networks,"
+              + " see Xpoa-block-txs-selection-max-time instead",
+          false,
+          singletonList("--Xblock-txs-selection-max-time"));
+    } else {
+      CommandLineUtils.failIfOptionDoesntMeetRequirement(
+          commandLine,
+          "--Xpoa-block-txs-selection-max-time can be only used with PoA networks,"
+              + " see --Xblock-txs-selection-max-time instead",
+          false,
+          singletonList("--Xpoa-block-txs-selection-max-time"));
+
+      if (unstableOptions.nonPoaBlockTxsSelectionMaxTime <= 0
+          || unstableOptions.nonPoaBlockTxsSelectionMaxTime
+              > DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME) {
+        throw new ParameterException(
+            commandLine,
+            "--Xblock-txs-selection-max-time must be positive and ≤ "
+                + DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME);
+      }
+    }
   }
 
   static MiningOptions fromConfig(final MiningParameters miningParameters) {
@@ -265,6 +317,10 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
         miningParameters.getUnstable().getPosBlockCreationMaxTime();
     miningOptions.unstableOptions.posBlockCreationRepetitionMinDuration =
         miningParameters.getUnstable().getPosBlockCreationRepetitionMinDuration();
+    miningOptions.unstableOptions.nonPoaBlockTxsSelectionMaxTime =
+        miningParameters.getUnstable().getBlockTxsSelectionMaxTime();
+    miningOptions.unstableOptions.poaBlockTxsSelectionMaxTime =
+        miningParameters.getUnstable().getPoaBlockTxsSelectionMaxTime();
 
     miningParameters.getCoinbase().ifPresent(coinbase -> miningOptions.coinbase = coinbase);
     miningParameters.getTargetGasLimit().ifPresent(tgl -> miningOptions.targetGasLimit = tgl);
@@ -304,6 +360,8 @@ public class MiningOptions implements CLIOptions<MiningParameters> {
                     .posBlockCreationMaxTime(unstableOptions.posBlockCreationMaxTime)
                     .posBlockCreationRepetitionMinDuration(
                         unstableOptions.posBlockCreationRepetitionMinDuration)
+                    .nonPoaBlockTxsSelectionMaxTime(unstableOptions.nonPoaBlockTxsSelectionMaxTime)
+                    .poaBlockTxsSelectionMaxTime(unstableOptions.poaBlockTxsSelectionMaxTime)
                     .build());
 
     return miningParametersBuilder.build();

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

@@ -94,7 +94,8 @@ public class CliqueBesuControllerBuilder extends BesuControllerBuilder {
                 localAddress,
                 secondsBetweenBlocks),
             epochManager,
-            createEmptyBlocks);
+            createEmptyBlocks,
+            ethProtocolManager.ethContext().getScheduler());
     final CliqueMiningCoordinator miningCoordinator =
         new CliqueMiningCoordinator(
             protocolContext.getBlockchain(),

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

@@ -153,7 +153,8 @@ public class IbftBesuControllerBuilder extends BftBesuControllerBuilder {
             forksSchedule,
             miningParameters,
             localAddress,
-            bftExtraDataCodec().get());
+            bftExtraDataCodec().get(),
+            ethProtocolManager.ethContext().getScheduler());
 
     final ValidatorProvider validatorProvider =
         protocolContext.getConsensusContext(BftContext.class).getValidatorProvider();

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

@@ -55,7 +55,8 @@ public class MainnetBesuControllerBuilder extends BesuControllerBuilder {
                 MainnetBlockHeaderValidator.MINIMUM_SECONDS_SINCE_PARENT,
                 MainnetBlockHeaderValidator.TIMESTAMP_TOLERANCE_S,
                 clock),
-            epochCalculator);
+            epochCalculator,
+            ethProtocolManager.ethContext().getScheduler());
 
     final PoWMiningCoordinator miningCoordinator =
         new PoWMiningCoordinator(

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

@@ -35,7 +35,6 @@ import org.hyperledger.besu.ethereum.eth.manager.EthPeers;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.MergePeerFilter;
-import org.hyperledger.besu.ethereum.eth.manager.MonitoredExecutors;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.peervalidation.RequiredBlocksPeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
@@ -46,13 +45,10 @@ import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.mainnet.ScheduleBasedBlockHeaderFunctions;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
-import org.hyperledger.besu.plugin.services.MetricsSystem;
 
 import java.util.List;
 import java.util.Optional;
 import java.util.OptionalLong;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
@@ -85,7 +81,7 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
             syncState,
             BackwardChain.from(
                 storageProvider, ScheduleBasedBlockHeaderFunctions.create(protocolSchedule))),
-        metricsSystem);
+        ethProtocolManager.ethContext().getScheduler());
   }
 
   @Override
@@ -146,7 +142,7 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
    * @param miningParameters the mining parameters
    * @param syncState the sync state
    * @param backwardSyncContext the backward sync context
-   * @param metricsSystem the metrics system
+   * @param ethScheduler the scheduler
    * @return the mining coordinator
    */
   protected MiningCoordinator createTransitionMiningCoordinator(
@@ -156,13 +152,10 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
       final MiningParameters miningParameters,
       final SyncState syncState,
       final BackwardSyncContext backwardSyncContext,
-      final MetricsSystem metricsSystem) {
+      final EthScheduler ethScheduler) {
 
     this.syncState.set(syncState);
 
-    final ExecutorService blockBuilderExecutor =
-        MonitoredExecutors.newSingleThreadExecutor("PoS-Block-Builder", metricsSystem);
-
     final GenesisConfigOptions genesisConfigOptions = configOptionsSupplier.get();
     final Optional<Address> depositContractAddress =
         genesisConfigOptions.getDepositContractAddress();
@@ -170,10 +163,7 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
     return new MergeCoordinator(
         protocolContext,
         protocolSchedule,
-        task -> {
-          LOG.debug("Block builder executor status {}", blockBuilderExecutor);
-          return CompletableFuture.runAsync(task, blockBuilderExecutor);
-        },
+        ethScheduler,
         transactionPool,
         miningParameters,
         backwardSyncContext,

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

@@ -191,7 +191,8 @@ public class QbftBesuControllerBuilder extends BftBesuControllerBuilder {
             qbftForksSchedule,
             miningParameters,
             localAddress,
-            bftExtraDataCodec().get());
+            bftExtraDataCodec().get(),
+            ethProtocolManager.ethContext().getScheduler());
 
     final ValidatorProvider validatorProvider =
         protocolContext.getConsensusContext(BftContext.class).getValidatorProvider();

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

@@ -144,7 +144,7 @@ public class TransitionBesuControllerBuilder extends BesuControllerBuilder {
                 transitionMiningParameters,
                 syncState,
                 transitionBackwardsSyncContext,
-                metricsSystem));
+                ethProtocolManager.ethContext().getScheduler()));
     initTransitionWatcher(protocolContext, composedCoordinator);
     return composedCoordinator;
   }

besu/src/test/java/org/hyperledger/besu/cli/options/MiningOptionsTest.java

@@ -15,6 +15,8 @@
 package org.hyperledger.besu.cli.options;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME;
 import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_POS_BLOCK_CREATION_MAX_TIME;
 import static org.mockito.Mockito.atMost;
 import static org.mockito.Mockito.verify;
@@ -26,6 +28,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitValues;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.Unstable;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.util.number.Percentage;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -308,6 +311,79 @@ public class MiningOptionsTest extends AbstractCLIOptionsTest<MiningParameters,
         "17000");
   }
 
+  @Test
+  public void blockTxsSelectionMaxTimeDefaultValue() {
+    internalTestSuccess(
+        miningParams ->
+            assertThat(miningParams.getUnstable().getBlockTxsSelectionMaxTime())
+                .isEqualTo(DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME));
+  }
+
+  @Test
+  public void blockTxsSelectionMaxTimeOption() {
+    internalTestSuccess(
+        miningParams ->
+            assertThat(miningParams.getUnstable().getBlockTxsSelectionMaxTime()).isEqualTo(1700L),
+        "--Xblock-txs-selection-max-time",
+        "1700");
+  }
+
+  @Test
+  public void blockTxsSelectionMaxTimeOutOfAllowedRange() {
+    internalTestFailure(
+        "--Xblock-txs-selection-max-time must be positive and ≤ 5000",
+        "--Xblock-txs-selection-max-time",
+        "6000");
+  }
+
+  @Test
+  public void blockTxsSelectionMaxTimeIncompatibleWithPoaNetworks() throws IOException {
+    final Path genesisFileIBFT2 = createFakeGenesisFile(VALID_GENESIS_IBFT2_POST_LONDON);
+    internalTestFailure(
+        "--Xblock-txs-selection-max-time can't be used with PoA networks, see Xpoa-block-txs-selection-max-time instead",
+        "--genesis-file",
+        genesisFileIBFT2.toString(),
+        "--Xblock-txs-selection-max-time",
+        "2");
+  }
+
+  @Test
+  public void poaBlockTxsSelectionMaxTimeDefaultValue() {
+    internalTestSuccess(
+        miningParams ->
+            assertThat(miningParams.getUnstable().getPoaBlockTxsSelectionMaxTime())
+                .isEqualTo(DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME));
+  }
+
+  @Test
+  public void poaBlockTxsSelectionMaxTimeOption() throws IOException {
+    final Path genesisFileIBFT2 = createFakeGenesisFile(VALID_GENESIS_IBFT2_POST_LONDON);
+    internalTestSuccess(
+        miningParams ->
+            assertThat(miningParams.getUnstable().getPoaBlockTxsSelectionMaxTime())
+                .isEqualTo(Percentage.fromInt(80)),
+        "--genesis-file",
+        genesisFileIBFT2.toString(),
+        "--Xpoa-block-txs-selection-max-time",
+        "80");
+  }
+
+  @Test
+  public void poaBlockTxsSelectionMaxTimeOutOfAllowedRange() {
+    internalTestFailure(
+        "Invalid value for option '--Xpoa-block-txs-selection-max-time': cannot convert '110' to Percentage",
+        "--Xpoa-block-txs-selection-max-time",
+        "110");
+  }
+
+  @Test
+  public void poaBlockTxsSelectionMaxTimeOnlyCompatibleWithPoaNetworks() {
+    internalTestFailure(
+        "--Xpoa-block-txs-selection-max-time can be only used with PoA networks, see --Xblock-txs-selection-max-time instead",
+        "--Xpoa-block-txs-selection-max-time",
+        "90");
+  }
+
   @Override
   protected MiningParameters createDefaultDomainObject() {
     return MiningParameters.newDefault();

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

@@ -49,12 +49,14 @@ import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
 import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.util.Optional;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Answers;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 
@@ -71,7 +73,10 @@ public class TransitionControllerBuilderTest {
   @Mock MutableBlockchain mockBlockchain;
   @Mock TransactionPool transactionPool;
   @Mock SyncState syncState;
-  @Mock EthProtocolManager ethProtocolManager;
+
+  @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+  EthProtocolManager ethProtocolManager;
+
   @Mock PostMergeContext mergeContext;
   StorageProvider storageProvider = new InMemoryKeyValueStorageProvider();
 
@@ -101,6 +106,8 @@ public class TransitionControllerBuilderTest {
         .thenReturn(mock(CliqueContext.class));
     when(protocolContext.getConsensusContext(PostMergeContext.class)).thenReturn(mergeContext);
     when(protocolContext.getConsensusContext(MergeContext.class)).thenReturn(mergeContext);
+    when(ethProtocolManager.ethContext().getScheduler())
+        .thenReturn(new DeterministicEthScheduler());
     miningParameters = MiningParameters.newDefault();
   }
 

consensus/clique/src/main/java/org/hyperledger/besu/consensus/clique/blockcreation/CliqueBlockCreator.java

@@ -32,6 +32,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeaderFunctions;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Util;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.mainnet.ScheduleBasedBlockHeaderFunctions;
@@ -55,6 +56,7 @@ public class CliqueBlockCreator extends AbstractBlockCreator {
    * @param nodeKey the node key
    * @param parentHeader the parent header
    * @param epochManager the epoch manager
+   * @param ethScheduler the scheduler for asynchronous block creation tasks
    */
   public CliqueBlockCreator(
       final MiningParameters miningParameters,
@@ -64,7 +66,8 @@ public class CliqueBlockCreator extends AbstractBlockCreator {
       final ProtocolSchedule protocolSchedule,
       final NodeKey nodeKey,
       final BlockHeader parentHeader,
-      final EpochManager epochManager) {
+      final EpochManager epochManager,
+      final EthScheduler ethScheduler) {
     super(
         miningParameters,
         __ -> Util.publicKeyToAddress(nodeKey.getPublicKey()),
@@ -73,7 +76,8 @@ public class CliqueBlockCreator extends AbstractBlockCreator {
         protocolContext,
         protocolSchedule,
         parentHeader,
-        Optional.empty());
+        Optional.empty(),
+        ethScheduler);
     this.nodeKey = nodeKey;
     this.epochManager = epochManager;
   }

consensus/clique/src/main/java/org/hyperledger/besu/consensus/clique/blockcreation/CliqueMinerExecutor.java

@@ -28,6 +28,7 @@ import org.hyperledger.besu.ethereum.chain.PoWObserver;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.Util;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.util.Subscribers;
@@ -60,6 +61,7 @@ public class CliqueMinerExecutor extends AbstractMinerExecutor<CliqueBlockMiner>
    * @param blockScheduler the block scheduler
    * @param epochManager the epoch manager
    * @param createEmptyBlocks whether clique should allow the creation of empty blocks.
+   * @param ethScheduler the scheduler for asynchronous block creation tasks
    */
   public CliqueMinerExecutor(
       final ProtocolContext protocolContext,
@@ -69,8 +71,15 @@ public class CliqueMinerExecutor extends AbstractMinerExecutor<CliqueBlockMiner>
       final MiningParameters miningParams,
       final AbstractBlockScheduler blockScheduler,
       final EpochManager epochManager,
-      final boolean createEmptyBlocks) {
-    super(protocolContext, protocolSchedule, transactionPool, miningParams, blockScheduler);
+      final boolean createEmptyBlocks,
+      final EthScheduler ethScheduler) {
+    super(
+        protocolContext,
+        protocolSchedule,
+        transactionPool,
+        miningParams,
+        blockScheduler,
+        ethScheduler);
     this.nodeKey = nodeKey;
     this.localAddress = Util.publicKeyToAddress(nodeKey.getPublicKey());
     this.epochManager = epochManager;
@@ -93,7 +102,8 @@ public class CliqueMinerExecutor extends AbstractMinerExecutor<CliqueBlockMiner>
                 protocolSchedule,
                 nodeKey,
                 header,
-                epochManager);
+                epochManager,
+                ethScheduler);
 
     return new CliqueBlockMiner(
         blockCreator,

consensus/clique/src/test/java/org/hyperledger/besu/consensus/clique/blockcreation/CliqueBlockCreatorTest.java

@@ -52,6 +52,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitV
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -63,6 +64,7 @@ import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 
 import java.time.ZoneId;
@@ -83,7 +85,7 @@ public class CliqueBlockCreatorTest {
   private final List<Address> validatorList = Lists.newArrayList();
   private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
   private final CliqueBlockInterface blockInterface = new CliqueBlockInterface();
-
+  private final EthScheduler ethScheduler = new DeterministicEthScheduler();
   private ProtocolSchedule protocolSchedule;
   private final WorldStateArchive stateArchive = createInMemoryWorldStateArchive();
 
@@ -148,7 +150,8 @@ public class CliqueBlockCreatorTest {
             protocolSchedule,
             proposerNodeKey,
             blockchain.getChainHeadHeader(),
-            epochManager);
+            epochManager,
+            ethScheduler);
 
     final Block createdBlock = blockCreator.createBlock(5L).getBlock();
 
@@ -176,7 +179,8 @@ public class CliqueBlockCreatorTest {
             protocolSchedule,
             proposerNodeKey,
             blockchain.getChainHeadHeader(),
-            epochManager);
+            epochManager,
+            ethScheduler);
 
     final Block createdBlock = blockCreator.createBlock(0L).getBlock();
     assertThat(createdBlock.getHeader().getNonce()).isEqualTo(CliqueBlockInterface.ADD_NONCE);
@@ -209,7 +213,8 @@ public class CliqueBlockCreatorTest {
             protocolSchedule,
             proposerNodeKey,
             blockchain.getChainHeadHeader(),
-            epochManager);
+            epochManager,
+            ethScheduler);
 
     final Block createdBlock = blockCreator.createBlock(0L).getBlock();
     assertThat(createdBlock.getHeader().getNonce()).isEqualTo(CliqueBlockInterface.DROP_NONCE);

consensus/clique/src/test/java/org/hyperledger/besu/consensus/clique/blockcreation/CliqueMinerExecutorTest.java

@@ -42,6 +42,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitV
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -51,6 +52,7 @@ import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 
 import java.time.ZoneId;
@@ -78,6 +80,7 @@ public class CliqueMinerExecutorTest {
   private BlockHeaderTestFixture blockHeaderBuilder;
   private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
   private final CliqueBlockInterface blockInterface = new CliqueBlockInterface();
+  private final EthScheduler ethScheduler = new DeterministicEthScheduler();
 
   @BeforeEach
   public void setup() {
@@ -114,7 +117,8 @@ public class CliqueMinerExecutorTest {
             miningParameters,
             mock(CliqueBlockScheduler.class),
             new EpochManager(EPOCH_LENGTH),
-            true);
+            true,
+            ethScheduler);
 
     // NOTE: Passing in the *parent* block, so must be 1 less than EPOCH
     final BlockHeader header = blockHeaderBuilder.number(EPOCH_LENGTH - 1).buildHeader();
@@ -149,7 +153,8 @@ public class CliqueMinerExecutorTest {
             miningParameters,
             mock(CliqueBlockScheduler.class),
             new EpochManager(EPOCH_LENGTH),
-            true);
+            true,
+            ethScheduler);
 
     // Parent block was epoch, so the next block should contain no validators.
     final BlockHeader header = blockHeaderBuilder.number(EPOCH_LENGTH).buildHeader();
@@ -184,7 +189,8 @@ public class CliqueMinerExecutorTest {
             miningParameters,
             mock(CliqueBlockScheduler.class),
             new EpochManager(EPOCH_LENGTH),
-            true);
+            true,
+            ethScheduler);
 
     executor.setExtraData(modifiedVanityData);
     final Bytes extraDataBytes = executor.calculateExtraData(blockHeaderBuilder.buildHeader());

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

@@ -26,6 +26,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderBuilder;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 
@@ -50,6 +51,7 @@ public class BftBlockCreator extends AbstractBlockCreator {
    * @param protocolSchedule the protocol schedule
    * @param parentHeader the parent header
    * @param bftExtraDataCodec the bft extra data codec
+   * @param ethScheduler the scheduler for asynchronous block creation tasks
    */
   public BftBlockCreator(
       final MiningParameters miningParameters,
@@ -60,7 +62,8 @@ public class BftBlockCreator extends AbstractBlockCreator {
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
       final BlockHeader parentHeader,
-      final BftExtraDataCodec bftExtraDataCodec) {
+      final BftExtraDataCodec bftExtraDataCodec,
+      final EthScheduler ethScheduler) {
     super(
         miningParameters.setCoinbase(localAddress),
         miningBeneficiaryCalculator(localAddress, forksSchedule),
@@ -69,7 +72,8 @@ public class BftBlockCreator extends AbstractBlockCreator {
         protocolContext,
         protocolSchedule,
         parentHeader,
-        Optional.empty());
+        Optional.empty(),
+        ethScheduler);
     this.bftExtraDataCodec = bftExtraDataCodec;
   }
 

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

@@ -32,6 +32,7 @@ import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.blockcreation.BlockCreator;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.AbstractGasLimitSpecification;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
@@ -61,6 +62,8 @@ public class BftBlockCreatorFactory<T extends BftConfigOptions> {
   protected final ProtocolSchedule protocolSchedule;
   /** The Bft extra data codec. */
   protected final BftExtraDataCodec bftExtraDataCodec;
+  /** The scheduler for asynchronous block creation tasks */
+  protected final EthScheduler ethScheduler;
 
   private final Address localAddress;
 
@@ -74,6 +77,7 @@ public class BftBlockCreatorFactory<T extends BftConfigOptions> {
    * @param miningParams the mining params
    * @param localAddress the local address
    * @param bftExtraDataCodec the bft extra data codec
+   * @param ethScheduler the scheduler for asynchronous block creation tasks
    */
   public BftBlockCreatorFactory(
       final TransactionPool transactionPool,
@@ -82,7 +86,8 @@ public class BftBlockCreatorFactory<T extends BftConfigOptions> {
       final ForksSchedule<T> forksSchedule,
       final MiningParameters miningParams,
       final Address localAddress,
-      final BftExtraDataCodec bftExtraDataCodec) {
+      final BftExtraDataCodec bftExtraDataCodec,
+      final EthScheduler ethScheduler) {
     this.transactionPool = transactionPool;
     this.protocolContext = protocolContext;
     this.protocolSchedule = protocolSchedule;
@@ -90,6 +95,7 @@ public class BftBlockCreatorFactory<T extends BftConfigOptions> {
     this.localAddress = localAddress;
     this.miningParameters = miningParams;
     this.bftExtraDataCodec = bftExtraDataCodec;
+    this.ethScheduler = ethScheduler;
   }
 
   /**
@@ -109,7 +115,8 @@ public class BftBlockCreatorFactory<T extends BftConfigOptions> {
         protocolContext,
         protocolSchedule,
         parentHeader,
-        bftExtraDataCodec);
+        bftExtraDataCodec,
+        ethScheduler);
   }
 
   /**

consensus/ibft/src/integration-test/java/org/hyperledger/besu/consensus/ibft/support/TestContextBuilder.java

@@ -82,6 +82,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitV
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -92,6 +93,7 @@ import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 import org.hyperledger.besu.util.Subscribers;
 
@@ -367,6 +369,8 @@ public class TestContextBuilder {
 
     transactionPool.setEnabled();
 
+    final EthScheduler ethScheduler = new DeterministicEthScheduler();
+
     final Address localAddress = Util.publicKeyToAddress(nodeKey.getPublicKey());
     final BftBlockCreatorFactory<?> blockCreatorFactory =
         new BftBlockCreatorFactory<>(
@@ -376,7 +380,8 @@ public class TestContextBuilder {
             forksSchedule,
             miningParams,
             localAddress,
-            IBFT_EXTRA_DATA_ENCODER);
+            IBFT_EXTRA_DATA_ENCODER,
+            ethScheduler);
 
     final ProposerSelector proposerSelector =
         new ProposerSelector(blockChain, blockInterface, true, validatorProvider);

consensus/ibft/src/test/java/org/hyperledger/besu/consensus/ibft/blockcreation/BftBlockCreatorTest.java

@@ -59,6 +59,7 @@ import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 
 import java.time.ZoneId;
@@ -186,7 +187,8 @@ public class BftBlockCreatorTest {
             protContext,
             protocolSchedule,
             parentHeader,
-            bftExtraDataEncoder);
+            bftExtraDataEncoder,
+            new DeterministicEthScheduler());
 
     final int secondsBetweenBlocks = 1;
     final Block block = blockCreator.createBlock(parentHeader.getTimestamp() + 1).getBlock();

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

@@ -24,6 +24,7 @@ import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.Withdrawal;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 
@@ -53,7 +54,8 @@ class MergeBlockCreator extends AbstractBlockCreator {
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
       final BlockHeader parentHeader,
-      final Optional<Address> depositContractAddress) {
+      final Optional<Address> depositContractAddress,
+      final EthScheduler ethScheduler) {
     super(
         miningParameters,
         __ -> miningParameters.getCoinbase().orElseThrow(),
@@ -62,7 +64,8 @@ class MergeBlockCreator extends AbstractBlockCreator {
         protocolContext,
         protocolSchedule,
         parentHeader,
-        depositContractAddress);
+        depositContractAddress,
+        ethScheduler);
   }
 
   /**

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

@@ -36,6 +36,7 @@ import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.MutableWorldState;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.Withdrawal;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BadChainListener;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -86,7 +87,7 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
   /** The Protocol context. */
   protected final ProtocolContext protocolContext;
   /** The Block builder executor. */
-  protected final ProposalBuilderExecutor blockBuilderExecutor;
+  protected final EthScheduler ethScheduler;
   /** The Backward sync context. */
   protected final BackwardSyncContext backwardSyncContext;
   /** The Protocol schedule. */
@@ -100,7 +101,7 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
    *
    * @param protocolContext the protocol context
    * @param protocolSchedule the protocol schedule
-   * @param blockBuilderExecutor the block builder executor
+   * @param ethScheduler the block builder executor
    * @param transactionPool the pending transactions
    * @param miningParams the mining params
    * @param backwardSyncContext the backward sync context
@@ -109,14 +110,14 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
   public MergeCoordinator(
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
-      final ProposalBuilderExecutor blockBuilderExecutor,
+      final EthScheduler ethScheduler,
       final TransactionPool transactionPool,
       final MiningParameters miningParams,
       final BackwardSyncContext backwardSyncContext,
       final Optional<Address> depositContractAddress) {
     this.protocolContext = protocolContext;
     this.protocolSchedule = protocolSchedule;
-    this.blockBuilderExecutor = blockBuilderExecutor;
+    this.ethScheduler = ethScheduler;
     this.mergeContext = protocolContext.getConsensusContext(MergeContext.class);
     this.backwardSyncContext = backwardSyncContext;
 
@@ -140,7 +141,8 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
               protocolContext,
               protocolSchedule,
               parentHeader,
-              depositContractAddress);
+              depositContractAddress,
+              ethScheduler);
         };
 
     this.backwardSyncContext.subscribeBadChainListener(this);
@@ -151,7 +153,7 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
    *
    * @param protocolContext the protocol context
    * @param protocolSchedule the protocol schedule
-   * @param blockBuilderExecutor the block builder executor
+   * @param ethScheduler the block builder executor
    * @param miningParams the mining params
    * @param backwardSyncContext the backward sync context
    * @param mergeBlockCreatorFactory the merge block creator factory
@@ -159,14 +161,14 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
   public MergeCoordinator(
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
-      final ProposalBuilderExecutor blockBuilderExecutor,
+      final EthScheduler ethScheduler,
       final MiningParameters miningParams,
       final BackwardSyncContext backwardSyncContext,
       final MergeBlockCreatorFactory mergeBlockCreatorFactory) {
 
     this.protocolContext = protocolContext;
     this.protocolSchedule = protocolSchedule;
-    this.blockBuilderExecutor = blockBuilderExecutor;
+    this.ethScheduler = ethScheduler;
     this.mergeContext = protocolContext.getConsensusContext(MergeContext.class);
     this.backwardSyncContext = backwardSyncContext;
     if (miningParams.getTargetGasLimit().isEmpty()) {
@@ -366,8 +368,9 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
         payloadIdentifier,
         miningParameters.getUnstable().getPosBlockCreationMaxTime());
 
-    blockBuilderExecutor
-        .buildProposal(() -> retryBlockCreationUntilUseful(payloadIdentifier, blockCreator))
+    ethScheduler
+        .scheduleBlockCreationTask(
+            () -> retryBlockCreationUntilUseful(payloadIdentifier, blockCreator))
         .orTimeout(
             miningParameters.getUnstable().getPosBlockCreationMaxTime(), TimeUnit.MILLISECONDS)
         .whenComplete(
@@ -895,15 +898,4 @@ public class MergeCoordinator implements MergeMiningCoordinator, BadChainListene
       blockCreator.cancel();
     }
   }
-
-  /** The interface Proposal builder executor. */
-  public interface ProposalBuilderExecutor {
-    /**
-     * Build proposal and return completable future.
-     *
-     * @param task the task
-     * @return the completable future
-     */
-    CompletableFuture<Void> buildProposal(final Runnable task);
-  }
 }

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeCoordinatorTest.java

@@ -35,7 +35,6 @@ import static org.mockito.Mockito.when;
 import org.hyperledger.besu.config.MergeConfigOptions;
 import org.hyperledger.besu.consensus.merge.MergeContext;
 import org.hyperledger.besu.consensus.merge.PayloadWrapper;
-import org.hyperledger.besu.consensus.merge.blockcreation.MergeCoordinator.ProposalBuilderExecutor;
 import org.hyperledger.besu.consensus.merge.blockcreation.MergeMiningCoordinator.ForkchoiceResult;
 import org.hyperledger.besu.crypto.KeyPair;
 import org.hyperledger.besu.crypto.SECPPrivateKey;
@@ -62,6 +61,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.Unstable;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.TransactionTestFixture;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
@@ -130,10 +130,11 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
   @Mock(answer = Answers.RETURNS_DEEP_STUBS)
   EthContext ethContext;
 
-  @Mock ProposalBuilderExecutor proposalBuilderExecutor;
+  @Mock EthScheduler ethScheduler;
+
   private final Address coinbase = genesisAllocations(getPosGenesisConfigFile()).findFirst().get();
 
-  MiningParameters miningParameters =
+  private MiningParameters miningParameters =
       ImmutableMiningParameters.builder()
           .mutableInitValues(MutableInitValues.builder().coinbase(coinbase).build())
           .unstable(
@@ -205,10 +206,13 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
     genesisState.writeStateTo(mutable);
     mutable.persist(null);
 
-    when(proposalBuilderExecutor.buildProposal(any()))
+    when(ethScheduler.scheduleBlockCreationTask(any()))
         .thenAnswer(
             invocation -> {
               final Runnable runnable = invocation.getArgument(0);
+              if (!invocation.toString().contains("MergeCoordinator")) {
+                return CompletableFuture.runAsync(runnable);
+              }
               blockCreationTask = CompletableFuture.runAsync(runnable);
               return blockCreationTask;
             });
@@ -234,7 +238,7 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
         new MergeCoordinator(
             protocolContext,
             protocolSchedule,
-            proposalBuilderExecutor,
+            ethScheduler,
             transactionPool,
             miningParameters,
             backwardSyncContext,
@@ -287,7 +291,8 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
                       protocolContext,
                       protocolSchedule,
                       parentHeader,
-                      Optional.empty()));
+                      Optional.empty(),
+                      ethScheduler));
 
           doCallRealMethod()
               .doCallRealMethod()
@@ -304,7 +309,7 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
             new MergeCoordinator(
                 protocolContext,
                 protocolSchedule,
-                proposalBuilderExecutor,
+                ethScheduler,
                 miningParameters,
                 backwardSyncContext,
                 mergeBlockCreatorFactory));
@@ -646,7 +651,7 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
     doAnswer(
             invocation -> {
               if (retries.getAndIncrement() < 5) {
-                // a new transaction every time a block is built
+                // add a new transaction every time a block is built
                 transactions.addTransaction(
                     createLocalTransaction(retries.get() - 1), Optional.empty());
               } else {
@@ -751,7 +756,7 @@ public class MergeCoordinatorTest implements MergeGenesisConfigHelper {
         new MergeCoordinator(
             protocolContext,
             protocolSchedule,
-            proposalBuilderExecutor,
+            ethScheduler,
             transactionPool,
             miningParameters,
             backwardSyncContext,

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeReorgTest.java

@@ -35,6 +35,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
 import org.hyperledger.besu.ethereum.core.Difficulty;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitValues;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.BlockHeaderValidator;
@@ -42,12 +43,12 @@ import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.BaseFeeMarket;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.LondonFeeMarket;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.util.LogConfigurator;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -72,7 +73,7 @@ public class MergeReorgTest implements MergeGenesisConfigHelper {
 
   private final WorldStateArchive worldStateArchive = createInMemoryWorldStateArchive();
   private final MutableBlockchain blockchain = createInMemoryBlockchain(genesisState.getBlock());
-
+  private final EthScheduler ethScheduler = new DeterministicEthScheduler();
   private final ProtocolContext protocolContext =
       new ProtocolContext(blockchain, worldStateArchive, mergeContext, Optional.empty());
 
@@ -92,7 +93,7 @@ public class MergeReorgTest implements MergeGenesisConfigHelper {
         new MergeCoordinator(
             protocolContext,
             mockProtocolSchedule,
-            CompletableFuture::runAsync,
+            ethScheduler,
             mockTransactionPool,
             ImmutableMiningParameters.builder()
                 .mutableInitValues(MutableInitValues.builder().coinbase(coinbase).build())

consensus/qbft/src/integration-test/java/org/hyperledger/besu/consensus/qbft/support/TestContextBuilder.java

@@ -96,6 +96,7 @@ import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.ProtocolScheduleFixture;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -108,6 +109,7 @@ import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 import org.hyperledger.besu.util.Subscribers;
 
@@ -455,6 +457,8 @@ public class TestContextBuilder {
 
     transactionPool.setEnabled();
 
+    final EthScheduler ethScheduler = new DeterministicEthScheduler();
+
     final Address localAddress = Util.publicKeyToAddress(nodeKey.getPublicKey());
     final BftBlockCreatorFactory<?> blockCreatorFactory =
         new QbftBlockCreatorFactory(
@@ -464,7 +468,8 @@ public class TestContextBuilder {
             forksSchedule,
             miningParams,
             localAddress,
-            BFT_EXTRA_DATA_ENCODER);
+            BFT_EXTRA_DATA_ENCODER,
+            ethScheduler);
 
     final ProposerSelector proposerSelector =
         new ProposerSelector(blockChain, blockInterface, true, validatorProvider);

consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/blockcreation/QbftBlockCreatorFactory.java

@@ -26,6 +26,7 @@ import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.blockcreation.BlockCreator;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 
@@ -46,6 +47,7 @@ public class QbftBlockCreatorFactory extends BftBlockCreatorFactory<QbftConfigOp
    * @param miningParams the mining params
    * @param localAddress the local address
    * @param bftExtraDataCodec the bft extra data codec
+   * @param ethScheduler the scheduler for asynchronous block creation tasks
    */
   public QbftBlockCreatorFactory(
       final TransactionPool transactionPool,
@@ -54,7 +56,8 @@ public class QbftBlockCreatorFactory extends BftBlockCreatorFactory<QbftConfigOp
       final ForksSchedule<QbftConfigOptions> forksSchedule,
       final MiningParameters miningParams,
       final Address localAddress,
-      final BftExtraDataCodec bftExtraDataCodec) {
+      final BftExtraDataCodec bftExtraDataCodec,
+      final EthScheduler ethScheduler) {
     super(
         transactionPool,
         protocolContext,
@@ -62,7 +65,8 @@ public class QbftBlockCreatorFactory extends BftBlockCreatorFactory<QbftConfigOp
         forksSchedule,
         miningParams,
         localAddress,
-        bftExtraDataCodec);
+        bftExtraDataCodec,
+        ethScheduler);
   }
 
   @Override

consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/blockcreation/QbftBlockCreatorFactoryTest.java

@@ -35,6 +35,7 @@ import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitV
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.util.Optional;
 
@@ -72,7 +73,8 @@ public class QbftBlockCreatorFactoryTest {
             forksSchedule,
             miningParams,
             mock(Address.class),
-            extraDataCodec);
+            extraDataCodec,
+            new DeterministicEthScheduler());
   }
 
   @Test

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/AbstractBlockCreator.java

@@ -39,6 +39,7 @@ import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.Withdrawal;
 import org.hyperledger.besu.ethereum.core.encoding.DepositDecoder;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.AbstractBlockProcessor;
 import org.hyperledger.besu.ethereum.mainnet.BodyValidation;
@@ -92,7 +93,7 @@ public abstract class AbstractBlockCreator implements AsyncBlockCreator {
   protected final BlockHeaderFunctions blockHeaderFunctions;
   protected final BlockHeader parentHeader;
   private final Optional<Address> depositContractAddress;
-
+  private final EthScheduler ethScheduler;
   private final AtomicBoolean isCancelled = new AtomicBoolean(false);
 
   protected AbstractBlockCreator(
@@ -103,7 +104,8 @@ public abstract class AbstractBlockCreator implements AsyncBlockCreator {
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
       final BlockHeader parentHeader,
-      final Optional<Address> depositContractAddress) {
+      final Optional<Address> depositContractAddress,
+      final EthScheduler ethScheduler) {
     this.miningParameters = miningParameters;
     this.miningBeneficiaryCalculator = miningBeneficiaryCalculator;
     this.extraDataCalculator = extraDataCalculator;
@@ -112,6 +114,7 @@ public abstract class AbstractBlockCreator implements AsyncBlockCreator {
     this.protocolSchedule = protocolSchedule;
     this.parentHeader = parentHeader;
     this.depositContractAddress = depositContractAddress;
+    this.ethScheduler = ethScheduler;
     blockHeaderFunctions = ScheduleBasedBlockHeaderFunctions.create(protocolSchedule);
   }
 
@@ -360,7 +363,8 @@ public abstract class AbstractBlockCreator implements AsyncBlockCreator {
             protocolSpec.getFeeMarket(),
             protocolSpec.getGasCalculator(),
             protocolSpec.getGasLimitCalculator(),
-            pluginTransactionSelector);
+            pluginTransactionSelector,
+            ethScheduler);
 
     if (transactions.isPresent()) {
       return selector.evaluateTransactions(transactions.get());

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/AbstractMinerExecutor.java

@@ -21,6 +21,7 @@ import org.hyperledger.besu.ethereum.chain.MinedBlockObserver;
 import org.hyperledger.besu.ethereum.chain.PoWObserver;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.AbstractGasLimitSpecification;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
@@ -41,12 +42,14 @@ public abstract class AbstractMinerExecutor<M extends BlockMiner<? extends Abstr
 
   private static final Logger LOG = LoggerFactory.getLogger(AbstractMinerExecutor.class);
 
-  private final ExecutorService executorService = Executors.newCachedThreadPool();
+  private final ExecutorService executorService =
+      Executors.newCachedThreadPool(r -> new Thread(r, "MinerExecutor"));
   protected final ProtocolContext protocolContext;
   protected final ProtocolSchedule protocolSchedule;
   protected final TransactionPool transactionPool;
   protected final AbstractBlockScheduler blockScheduler;
   protected final MiningParameters miningParameters;
+  protected final EthScheduler ethScheduler;
   private final AtomicBoolean stopped = new AtomicBoolean(false);
 
   protected AbstractMinerExecutor(
@@ -54,12 +57,14 @@ public abstract class AbstractMinerExecutor<M extends BlockMiner<? extends Abstr
       final ProtocolSchedule protocolSchedule,
       final TransactionPool transactionPool,
       final MiningParameters miningParams,
-      final AbstractBlockScheduler blockScheduler) {
+      final AbstractBlockScheduler blockScheduler,
+      final EthScheduler ethScheduler) {
     this.protocolContext = protocolContext;
     this.protocolSchedule = protocolSchedule;
     this.transactionPool = transactionPool;
     this.blockScheduler = blockScheduler;
     this.miningParameters = miningParams;
+    this.ethScheduler = ethScheduler;
   }
 
   public Optional<M> startAsyncMining(

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/PoWBlockCreator.java

@@ -19,6 +19,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderBuilder;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.EthHash;
 import org.hyperledger.besu.ethereum.mainnet.PoWSolution;
@@ -44,7 +45,8 @@ public class PoWBlockCreator extends AbstractBlockCreator {
       final ProtocolContext protocolContext,
       final ProtocolSchedule protocolSchedule,
       final PoWSolver nonceSolver,
-      final BlockHeader parentHeader) {
+      final BlockHeader parentHeader,
+      final EthScheduler ethScheduler) {
     super(
         miningParameters,
         __ -> miningParameters.getCoinbase().orElseThrow(),
@@ -53,7 +55,8 @@ public class PoWBlockCreator extends AbstractBlockCreator {
         protocolContext,
         protocolSchedule,
         parentHeader,
-        Optional.empty());
+        Optional.empty(),
+        ethScheduler);
 
     this.nonceSolver = nonceSolver;
   }

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/PoWMinerExecutor.java

@@ -20,6 +20,7 @@ import org.hyperledger.besu.ethereum.chain.MinedBlockObserver;
 import org.hyperledger.besu.ethereum.chain.PoWObserver;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.EpochCalculator;
 import org.hyperledger.besu.ethereum.mainnet.PoWSolver;
@@ -41,8 +42,15 @@ public class PoWMinerExecutor extends AbstractMinerExecutor<PoWBlockMiner> {
       final TransactionPool transactionPool,
       final MiningParameters miningParams,
       final AbstractBlockScheduler blockScheduler,
-      final EpochCalculator epochCalculator) {
-    super(protocolContext, protocolSchedule, transactionPool, miningParams, blockScheduler);
+      final EpochCalculator epochCalculator,
+      final EthScheduler ethScheduler) {
+    super(
+        protocolContext,
+        protocolSchedule,
+        transactionPool,
+        miningParams,
+        blockScheduler,
+        ethScheduler);
     if (miningParams.getNonceGenerator().isEmpty()) {
       miningParams.setNonceGenerator(new RandomNonceGenerator());
     }
@@ -85,7 +93,8 @@ public class PoWMinerExecutor extends AbstractMinerExecutor<PoWBlockMiner> {
                 protocolContext,
                 protocolSchedule,
                 solver,
-                parentHeader);
+                parentHeader,
+                ethScheduler);
 
     return new PoWBlockMiner(
         blockCreator, protocolSchedule, protocolContext, observers, blockScheduler, parentHeader);

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/txselection/BlockTransactionSelector.java

@@ -14,6 +14,9 @@
  */
 package org.hyperledger.besu.ethereum.blockcreation.txselection;
 
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.BLOCK_SELECTION_TIMEOUT;
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.SELECTED;
+
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.GasLimitCalculator;
@@ -29,6 +32,7 @@ import org.hyperledger.besu.ethereum.core.MutableWorldState;
 import org.hyperledger.besu.ethereum.core.ProcessableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.AbstractBlockProcessor;
@@ -46,6 +50,10 @@ import org.hyperledger.besu.plugin.services.txselection.PluginTransactionSelecto
 
 import java.util.List;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
 import org.slf4j.Logger;
@@ -85,6 +93,9 @@ public class BlockTransactionSelector {
   private final List<AbstractTransactionSelector> transactionSelectors;
   private final PluginTransactionSelector pluginTransactionSelector;
   private final BlockAwareOperationTracer pluginOperationTracer;
+  private final EthScheduler ethScheduler;
+  private final AtomicBoolean isTimeout = new AtomicBoolean(false);
+  private WorldUpdater blockWorldStateUpdater;
 
   public BlockTransactionSelector(
       final MiningParameters miningParameters,
@@ -100,12 +111,14 @@ public class BlockTransactionSelector {
       final FeeMarket feeMarket,
       final GasCalculator gasCalculator,
       final GasLimitCalculator gasLimitCalculator,
-      final PluginTransactionSelector pluginTransactionSelector) {
+      final PluginTransactionSelector pluginTransactionSelector,
+      final EthScheduler ethScheduler) {
     this.transactionProcessor = transactionProcessor;
     this.blockchain = blockchain;
     this.worldState = worldState;
     this.transactionReceiptFactory = transactionReceiptFactory;
     this.isCancelled = isCancelled;
+    this.ethScheduler = ethScheduler;
     this.blockSelectionContext =
         new BlockSelectionContext(
             miningParameters,
@@ -119,6 +132,7 @@ public class BlockTransactionSelector {
     transactionSelectors = createTransactionSelectors(blockSelectionContext);
     this.pluginTransactionSelector = pluginTransactionSelector;
     this.pluginOperationTracer = pluginTransactionSelector.getOperationTracer();
+    blockWorldStateUpdater = worldState.updater();
   }
 
   private List<AbstractTransactionSelector> createTransactionSelectors(
@@ -145,9 +159,7 @@ public class BlockTransactionSelector {
         .setMessage("Transaction pool stats {}")
         .addArgument(blockSelectionContext.transactionPool().logStats())
         .log();
-
-    blockSelectionContext.transactionPool().selectTransactions(this::evaluateTransaction);
-
+    timeLimitedSelection();
     LOG.atTrace()
         .setMessage("Transaction selection result {}")
         .addArgument(transactionSelectionResults::toTraceLog)
@@ -155,6 +167,36 @@ public class BlockTransactionSelector {
     return transactionSelectionResults;
   }
 
+  private void timeLimitedSelection() {
+    final long blockTxsSelectionMaxTime =
+        blockSelectionContext.miningParameters().getUnstable().getBlockTxsSelectionMaxTime();
+    final var txSelection =
+        ethScheduler.scheduleBlockCreationTask(
+            () ->
+                blockSelectionContext
+                    .transactionPool()
+                    .selectTransactions(this::evaluateTransaction));
+
+    try {
+      txSelection.get(blockTxsSelectionMaxTime, TimeUnit.MILLISECONDS);
+    } catch (InterruptedException | ExecutionException e) {
+      if (isCancelled.get()) {
+        throw new CancellationException("Cancelled during transaction selection");
+      }
+      LOG.warn("Error during block transaction selection", e);
+    } catch (TimeoutException e) {
+      // synchronize since we want to be sure that there is no concurrent state update
+      synchronized (isTimeout) {
+        isTimeout.set(true);
+      }
+      LOG.warn(
+          "Interrupting transaction selection since it is taking more than the max configured time of "
+              + blockTxsSelectionMaxTime
+              + "ms",
+          e);
+    }
+  }
+
   /**
    * Evaluates a list of transactions and updates the selection results accordingly. If a
    * transaction is not selected during the evaluation, it is updated as not selected in the
@@ -189,17 +231,18 @@ public class BlockTransactionSelector {
       return handleTransactionNotSelected(pendingTransaction, selectionResult);
     }
 
-    final WorldUpdater worldStateUpdater = worldState.updater();
+    final WorldUpdater txWorldStateUpdater = blockWorldStateUpdater.updater();
     final TransactionProcessingResult processingResult =
-        processTransaction(pendingTransaction, worldStateUpdater);
+        processTransaction(pendingTransaction, txWorldStateUpdater);
 
     var postProcessingSelectionResult =
         evaluatePostProcessing(pendingTransaction, processingResult);
-    if (!postProcessingSelectionResult.selected()) {
-      return handleTransactionNotSelected(pendingTransaction, postProcessingSelectionResult);
-    }
 
-    return handleTransactionSelected(pendingTransaction, processingResult, worldStateUpdater);
+    if (postProcessingSelectionResult.selected()) {
+      return handleTransactionSelected(pendingTransaction, processingResult, txWorldStateUpdater);
+    }
+    return handleTransactionNotSelected(
+        pendingTransaction, postProcessingSelectionResult, txWorldStateUpdater);
   }
 
   /**
@@ -218,7 +261,7 @@ public class BlockTransactionSelector {
       TransactionSelectionResult result =
           selector.evaluateTransactionPreProcessing(
               pendingTransaction, transactionSelectionResults);
-      if (!result.equals(TransactionSelectionResult.SELECTED)) {
+      if (!result.equals(SELECTED)) {
         return result;
       }
     }
@@ -243,7 +286,7 @@ public class BlockTransactionSelector {
       TransactionSelectionResult result =
           selector.evaluateTransactionPostProcessing(
               pendingTransaction, transactionSelectionResults, processingResult);
-      if (!result.equals(TransactionSelectionResult.SELECTED)) {
+      if (!result.equals(SELECTED)) {
         return result;
       }
     }
@@ -282,14 +325,13 @@ public class BlockTransactionSelector {
    *
    * @param pendingTransaction The pending transaction.
    * @param processingResult The result of the transaction processing.
-   * @param worldStateUpdater The world state updater.
+   * @param txWorldStateUpdater The world state updater.
    * @return The result of the transaction selection process.
    */
   private TransactionSelectionResult handleTransactionSelected(
       final PendingTransaction pendingTransaction,
       final TransactionProcessingResult processingResult,
-      final WorldUpdater worldStateUpdater) {
-    worldStateUpdater.commit();
+      final WorldUpdater txWorldStateUpdater) {
     final Transaction transaction = pendingTransaction.getTransaction();
 
     final long gasUsedByTransaction =
@@ -299,17 +341,47 @@ public class BlockTransactionSelector {
     final long blobGasUsed =
         blockSelectionContext.gasCalculator().blobGasCost(transaction.getBlobCount());
 
+    final boolean tooLate;
+
+    // only add this tx to the selected set if it is not too late,
+    // this need to be done synchronously to avoid that a concurrent timeout
+    // could start packing a block while we are updating the state here
+    synchronized (isTimeout) {
+      if (!isTimeout.get()) {
+        txWorldStateUpdater.commit();
+        blockWorldStateUpdater.commit();
         final TransactionReceipt receipt =
             transactionReceiptFactory.create(
                 transaction.getType(), processingResult, worldState, cumulativeGasUsed);
 
-    logTransactionSelection(pendingTransaction.getTransaction());
-
         transactionSelectionResults.updateSelected(
             pendingTransaction.getTransaction(), receipt, gasUsedByTransaction, blobGasUsed);
-    pluginTransactionSelector.onTransactionSelected(pendingTransaction, processingResult);
+        tooLate = false;
+      } else {
+        tooLate = true;
+      }
+    }
+
+    if (tooLate) {
+      // even if this tx passed all the checks, it is too late to include it in this block,
+      // so we need to treat it as not selected
+      LOG.atTrace()
+          .setMessage("{} processed too late for block creation")
+          .addArgument(transaction::toTraceLog)
+          .log();
+      // do not rely on the presence of this result, since by the time it is added, the code
+      // reading it could have been already executed by another thread
+      return handleTransactionNotSelected(
+          pendingTransaction, BLOCK_SELECTION_TIMEOUT, txWorldStateUpdater);
+    }
 
-    return TransactionSelectionResult.SELECTED;
+    pluginTransactionSelector.onTransactionSelected(pendingTransaction, processingResult);
+    blockWorldStateUpdater = worldState.updater();
+    LOG.atTrace()
+        .setMessage("Selected {} for block creation")
+        .addArgument(transaction::toTraceLog)
+        .log();
+    return SELECTED;
   }
 
   /**
@@ -324,22 +396,24 @@ public class BlockTransactionSelector {
   private TransactionSelectionResult handleTransactionNotSelected(
       final PendingTransaction pendingTransaction,
       final TransactionSelectionResult selectionResult) {
+
     transactionSelectionResults.updateNotSelected(
         pendingTransaction.getTransaction(), selectionResult);
     pluginTransactionSelector.onTransactionNotSelected(pendingTransaction, selectionResult);
     return selectionResult;
   }
 
+  private TransactionSelectionResult handleTransactionNotSelected(
+      final PendingTransaction pendingTransaction,
+      final TransactionSelectionResult selectionResult,
+      final WorldUpdater txWorldStateUpdater) {
+    txWorldStateUpdater.revert();
+    return handleTransactionNotSelected(pendingTransaction, selectionResult);
+  }
+
   private void checkCancellation() {
     if (isCancelled.get()) {
       throw new CancellationException("Cancelled during transaction selection.");
     }
   }
-
-  private void logTransactionSelection(final Transaction transaction) {
-    LOG.atTrace()
-        .setMessage("Selected {} for block creation")
-        .addArgument(transaction::toTraceLog)
-        .log();
-  }
 }

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/txselection/TransactionSelectionResults.java

@@ -21,10 +21,10 @@ import org.hyperledger.besu.plugin.data.TransactionSelectionResult;
 
 import java.util.ArrayList;
 import java.util.EnumMap;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -39,8 +39,13 @@ public class TransactionSelectionResults {
   private final Map<TransactionType, List<Transaction>> transactionsByType =
       new EnumMap<>(TransactionType.class);
   private final List<TransactionReceipt> receipts = Lists.newArrayList();
+  /**
+   * Access to this field needs to be guarded, since it is possible to read it while another
+   * processing thread is writing, when the selection time is over.
+   */
   private final Map<Transaction, TransactionSelectionResult> notSelectedTransactions =
-      new HashMap<>();
+      new ConcurrentHashMap<>();
+
   private long cumulativeGasUsed = 0;
   private long cumulativeBlobGasUsed = 0;
 
@@ -92,18 +97,19 @@ public class TransactionSelectionResults {
   }
 
   public Map<Transaction, TransactionSelectionResult> getNotSelectedTransactions() {
-    return notSelectedTransactions;
+    return Map.copyOf(notSelectedTransactions);
   }
 
   public void logSelectionStats() {
     if (LOG.isDebugEnabled()) {
+      final var notSelectedTxs = getNotSelectedTransactions();
       final Map<TransactionSelectionResult, Long> notSelectedStats =
-          notSelectedTransactions.values().stream()
+          notSelectedTxs.values().stream()
               .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
 
       LOG.debug(
           "Selection stats: Totals[Evaluated={}, Selected={}, NotSelected={}, Discarded={}]; Detailed[{}]",
-          selectedTransactions.size() + notSelectedTransactions.size(),
+          selectedTransactions.size() + notSelectedTxs.size(),
           selectedTransactions.size(),
           notSelectedStats.size(),
           notSelectedStats.entrySet().stream()

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/AbstractBlockCreatorTest.java

@@ -59,6 +59,7 @@ import org.hyperledger.besu.ethereum.core.TransactionReceipt;
 import org.hyperledger.besu.ethereum.core.TransactionTestFixture;
 import org.hyperledger.besu.ethereum.core.Withdrawal;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -77,6 +78,7 @@ import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.evm.log.Log;
 import org.hyperledger.besu.evm.log.LogTopic;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.math.BigInteger;
 import java.time.Clock;
@@ -97,6 +99,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
 abstract class AbstractBlockCreatorTest {
   private static final Optional<Address> EMPTY_DEPOSIT_CONTRACT_ADDRESS = Optional.empty();
   @Mock private WithdrawalsProcessor withdrawalsProcessor;
+  protected EthScheduler ethScheduler = new DeterministicEthScheduler();
 
   @Test
   void findDepositsFromReceipts() {
@@ -405,7 +408,8 @@ abstract class AbstractBlockCreatorTest {
         executionContextTestFixture.getProtocolContext(),
         executionContextTestFixture.getProtocolSchedule(),
         blockchain.getChainHeadHeader(),
-        depositContractAddress);
+        depositContractAddress,
+        ethScheduler);
   }
 
   static class TestBlockCreator extends AbstractBlockCreator {
@@ -418,7 +422,8 @@ abstract class AbstractBlockCreatorTest {
         final ProtocolContext protocolContext,
         final ProtocolSchedule protocolSchedule,
         final BlockHeader parentHeader,
-        final Optional<Address> depositContractAddress) {
+        final Optional<Address> depositContractAddress,
+        final EthScheduler ethScheduler) {
       super(
           miningParameters,
           miningBeneficiaryCalculator,
@@ -427,7 +432,8 @@ abstract class AbstractBlockCreatorTest {
           protocolContext,
           protocolSchedule,
           parentHeader,
-          depositContractAddress);
+          depositContractAddress,
+          ethScheduler);
     }
 
     @Override

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/AbstractBlockTransactionSelectorTest.java

@@ -16,6 +16,8 @@ package org.hyperledger.besu.ethereum.blockcreation;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
+import static org.awaitility.Awaitility.await;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.eq;
@@ -48,6 +50,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
 import org.hyperledger.besu.ethereum.core.Difficulty;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
 import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.MutableInitValues;
+import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters.Unstable;
 import org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.MutableWorldState;
@@ -56,6 +59,7 @@ import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
 import org.hyperledger.besu.ethereum.difficulty.fixed.FixedDifficultyProtocolSchedule;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.MainnetBlockHeaderFunctions;
 import org.hyperledger.besu.ethereum.mainnet.MainnetTransactionProcessor;
@@ -76,6 +80,7 @@ import org.hyperledger.besu.plugin.services.MetricsSystem;
 import org.hyperledger.besu.plugin.services.txselection.PluginTransactionSelector;
 import org.hyperledger.besu.plugin.services.txselection.PluginTransactionSelectorFactory;
 import org.hyperledger.besu.services.kvstore.InMemoryKeyValueStorage;
+import org.hyperledger.besu.util.number.Percentage;
 
 import java.math.BigInteger;
 import java.time.Instant;
@@ -83,24 +88,34 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import com.google.common.collect.Lists;
 import org.apache.tuweni.bytes.Bytes;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.Answers;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
+import org.mockito.stubbing.Answer;
 
 @ExtendWith(MockitoExtension.class)
 @MockitoSettings(strictness = Strictness.LENIENT)
 public abstract class AbstractBlockTransactionSelectorTest {
   protected static final double MIN_OCCUPANCY_80_PERCENT = 0.8;
   protected static final double MIN_OCCUPANCY_100_PERCENT = 1;
+  protected static final PluginTransactionSelectorFactory NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY =
+      () -> AllAcceptingTransactionSelector.INSTANCE;
   protected static final BigInteger CHAIN_ID = BigInteger.valueOf(42L);
   protected static final KeyPair keyPair =
       SignatureAlgorithmFactory.getInstance().generateKeyPair();
@@ -113,7 +128,11 @@ public abstract class AbstractBlockTransactionSelectorTest {
   protected TransactionPool transactionPool;
   protected MutableWorldState worldState;
   protected ProtocolSchedule protocolSchedule;
-  protected MiningParameters miningParameters;
+  protected final MiningParameters defaultTestMiningParameters =
+      createMiningParameters(
+          Wei.ZERO, MIN_OCCUPANCY_80_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME);
+
+  @Mock protected EthScheduler ethScheduler;
 
   @Mock(answer = Answers.RETURNS_DEEP_STUBS)
   protected ProtocolContext protocolContext;
@@ -150,19 +169,15 @@ public abstract class AbstractBlockTransactionSelectorTest {
     when(protocolContext.getWorldStateArchive().getMutable(any(), anyBoolean()))
         .thenReturn(Optional.of(worldState));
     when(ethContext.getEthPeers().subscribeConnect(any())).thenReturn(1L);
-    miningParameters =
-        ImmutableMiningParameters.builder()
-            .mutableInitValues(MutableInitValues.builder().minTransactionGasPrice(Wei.ONE).build())
-            .build();
-
-    transactionPool = createTransactionPool();
+    when(ethScheduler.scheduleBlockCreationTask(any(Runnable.class)))
+        .thenAnswer(invocation -> CompletableFuture.runAsync(invocation.getArgument(0)));
   }
 
   protected abstract GenesisConfigFile getGenesisConfigFile();
 
   protected abstract ProtocolSchedule createProtocolSchedule();
 
-  protected abstract TransactionPool createTransactionPool();
+  protected abstract TransactionPool createTransactionPool(final MiningParameters miningParameters);
 
   private Boolean isCancelled() {
     return false;
@@ -198,13 +213,13 @@ public abstract class AbstractBlockTransactionSelectorTest {
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
 
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             mainnetTransactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -216,23 +231,21 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
   @Test
   public void validPendingTransactionIsIncludedInTheBlock() {
-    final Transaction transaction = createTransaction(1, Wei.of(7L), 100_000);
-    transactionPool.addRemoteTransactions(List.of(transaction));
-
-    ensureTransactionIsValid(transaction, 0, 5);
-
     final ProcessableBlockHeader blockHeader = createBlock(500_000);
-
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
-
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    final Transaction transaction = createTransaction(1, Wei.of(7L), 100_000);
+    transactionPool.addRemoteTransactions(List.of(transaction));
+
+    ensureTransactionIsValid(transaction, 0, 5);
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -244,6 +257,18 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
   @Test
   public void invalidTransactionsAreSkippedButBlockStillFills() {
+    // The block should fit 4 transactions only
+    final ProcessableBlockHeader blockHeader = createBlock(400_000);
+    final Address miningBeneficiary = AddressHelpers.ofValue(1);
+    final BlockTransactionSelector selector =
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
+            transactionProcessor,
+            blockHeader,
+            miningBeneficiary,
+            Wei.ZERO,
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
     final List<Transaction> transactionsToInject = Lists.newArrayList();
     for (int i = 0; i < 5; i++) {
       final Transaction tx = createTransaction(i, Wei.of(7), 100_000);
@@ -256,20 +281,6 @@ public abstract class AbstractBlockTransactionSelectorTest {
     }
     transactionPool.addRemoteTransactions(transactionsToInject);
 
-    // The block should fit 4 transactions only
-    final ProcessableBlockHeader blockHeader = createBlock(400_000);
-
-    final Address miningBeneficiary = AddressHelpers.ofValue(1);
-
-    final BlockTransactionSelector selector =
-        createBlockSelector(
-            transactionProcessor,
-            blockHeader,
-            Wei.ZERO,
-            miningBeneficiary,
-            Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
-
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
     final Transaction invalidTx = transactionsToInject.get(1);
@@ -288,26 +299,24 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
   @Test
   public void subsetOfPendingTransactionsIncludedWhenBlockGasLimitHit() {
-    final List<Transaction> transactionsToInject = Lists.newArrayList();
-    for (int i = 0; i < 5; i++) {
-      final Transaction tx = createTransaction(i, Wei.of(7), 100_000);
-      transactionsToInject.add(tx);
-      ensureTransactionIsValid(tx);
-    }
-    transactionPool.addRemoteTransactions(transactionsToInject);
-
     final ProcessableBlockHeader blockHeader = createBlock(301_000);
-
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
-
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    final List<Transaction> transactionsToInject = Lists.newArrayList();
+    for (int i = 0; i < 5; i++) {
+      final Transaction tx = createTransaction(i, Wei.of(7), 100_000);
+      transactionsToInject.add(tx);
+      ensureTransactionIsValid(tx);
+    }
+    transactionPool.addRemoteTransactions(transactionsToInject);
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -332,16 +341,15 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void transactionTooLargeForBlockDoesNotPreventMoreBeingAddedIfBlockOccupancyNotReached() {
     final ProcessableBlockHeader blockHeader = createBlock(300_000);
-
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     // Add 3 transactions to the Pending Transactions, 79% of block, 100% of block and 10% of block
     // should end up selecting the first and third only.
@@ -368,16 +376,15 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void transactionSelectionStopsWhenSufficientBlockOccupancyIsReached() {
     final ProcessableBlockHeader blockHeader = createBlock(300_000);
-
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     // Add 4 transactions to the Pending Transactions 15% (ok), 79% (ok), 25% (too large), 10%
     // (not included, it would fit, however previous transaction was too large and block was
@@ -409,13 +416,14 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.ZERO, MIN_OCCUPANCY_100_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_100_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     final long minTxGasCost = getGasCalculator().getMinimumTransactionCost();
 
@@ -467,13 +475,14 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.ZERO, MIN_OCCUPANCY_100_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_100_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     final long minTxGasCost = getGasCalculator().getMinimumTransactionCost();
 
@@ -519,13 +528,13 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     final Transaction validTransaction = createTransaction(0, Wei.of(10), 21_000);
 
@@ -552,6 +561,7 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void transactionSelectionPluginShouldWork_PreProcessing() {
     final ProcessableBlockHeader blockHeader = createBlock(300_000);
+    final Address miningBeneficiary = AddressHelpers.ofValue(1);
 
     final Transaction selected = createTransaction(0, Wei.of(10), 21_000);
     ensureTransactionIsValid(selected, 21_000, 0);
@@ -584,9 +594,9 @@ public abstract class AbstractBlockTransactionSelectorTest {
               }
             };
 
-    final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelectorWithTxSelPlugin(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
             miningBeneficiary,
@@ -648,7 +658,9 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelectorWithTxSelPlugin(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.ZERO, MIN_OCCUPANCY_80_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
             miningBeneficiary,
@@ -678,15 +690,19 @@ public abstract class AbstractBlockTransactionSelectorTest {
     final TransactionInvalidReason invalidReason = TransactionInvalidReason.PLUGIN_TX_VALIDATOR;
     final Transaction invalidTransaction = createTransaction(1, Wei.of(10), 21_000);
     ensureTransactionIsInvalid(invalidTransaction, TransactionInvalidReason.PLUGIN_TX_VALIDATOR);
-    transactionPool.addRemoteTransactions(List.of(transaction, invalidTransaction));
 
-    createBlockSelectorWithTxSelPlugin(
+    final BlockTransactionSelector selector =
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             createBlock(300_000),
             AddressHelpers.ofValue(1),
             Wei.ZERO,
-            transactionSelectorFactory)
-        .buildTransactionListForBlock();
+            transactionSelectorFactory);
+
+    transactionPool.addRemoteTransactions(List.of(transaction, invalidTransaction));
+
+    selector.buildTransactionListForBlock();
 
     ArgumentCaptor<PendingTransaction> argumentCaptor =
         ArgumentCaptor.forClass(PendingTransaction.class);
@@ -709,21 +725,20 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void transactionWithIncorrectNonceRemainsInPoolAndNotSelected() {
     final ProcessableBlockHeader blockHeader = createBlock(5_000_000);
-
-    final Transaction futureTransaction = createTransaction(4, Wei.of(10), 100_000);
-
-    transactionPool.addRemoteTransactions(List.of(futureTransaction));
-    ensureTransactionIsInvalid(futureTransaction, TransactionInvalidReason.NONCE_TOO_HIGH);
-
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    final Transaction futureTransaction = createTransaction(4, Wei.of(10), 100_000);
+
+    transactionPool.addRemoteTransactions(List.of(futureTransaction));
+    ensureTransactionIsInvalid(futureTransaction, TransactionInvalidReason.NONCE_TOO_HIGH);
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -740,22 +755,25 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void increaseOfMinGasPriceAtRuntimeExcludeTxFromBeingSelected() {
     final Transaction transaction = createTransaction(0, Wei.of(7L), 100_000);
-    transactionPool.addRemoteTransactions(List.of(transaction));
-
-    ensureTransactionIsValid(transaction, 0, 5);
-
     final ProcessableBlockHeader blockHeader = createBlock(500_000);
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
 
+    final MiningParameters miningParameters =
+        ImmutableMiningParameters.builder().from(defaultTestMiningParameters).build();
+
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            miningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    transactionPool.addRemoteTransactions(List.of(transaction));
+
+    ensureTransactionIsValid(transaction, 0, 5);
 
     // raise the minGasPrice at runtime from 1 wei to 10 wei
     miningParameters.setMinTransactionGasPrice(Wei.of(10));
@@ -774,22 +792,23 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void decreaseOfMinGasPriceAtRuntimeIncludeTxThatWasPreviouslyNotSelected() {
     final Transaction transaction = createTransaction(0, Wei.of(7L), 100_000);
-    transactionPool.addRemoteTransactions(List.of(transaction));
-
-    ensureTransactionIsValid(transaction, 0, 5);
-
+    final MiningParameters miningParameters =
+        ImmutableMiningParameters.builder().from(defaultTestMiningParameters).build();
     final ProcessableBlockHeader blockHeader = createBlock(500_000);
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
 
     final BlockTransactionSelector selector1 =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            miningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+    transactionPool.addRemoteTransactions(List.of(transaction));
+
+    ensureTransactionIsValid(transaction, 0, 5);
 
     // raise the minGasPrice at runtime from 1 wei to 10 wei
     miningParameters.setMinTransactionGasPrice(Wei.of(10));
@@ -809,12 +828,12 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
     final BlockTransactionSelector selector2 =
         createBlockSelector(
+            miningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     final TransactionSelectionResults results2 = selector2.buildTransactionListForBlock();
 
@@ -826,22 +845,25 @@ public abstract class AbstractBlockTransactionSelectorTest {
   @Test
   public void shouldNotSelectTransactionsWithPriorityFeeLessThanConfig() {
     ProcessableBlockHeader blockHeader = createBlock(5_000_000, Wei.ONE);
+    final MiningParameters miningParameters =
+        ImmutableMiningParameters.builder().from(defaultTestMiningParameters).build();
     miningParameters.setMinPriorityFeePerGas(Wei.of(7));
     final Transaction txSelected = createTransaction(1, Wei.of(8), 100_000);
     ensureTransactionIsValid(txSelected);
     // transaction txNotSelected should not be selected
     final Transaction txNotSelected = createTransaction(2, Wei.of(7), 100_000);
     ensureTransactionIsValid(txNotSelected);
-    transactionPool.addRemoteTransactions(List.of(txSelected, txNotSelected));
 
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            miningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             AddressHelpers.ofValue(1),
             Wei.ZERO,
-            MIN_OCCUPANCY_100_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    transactionPool.addRemoteTransactions(List.of(txSelected, txNotSelected));
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -852,41 +874,136 @@ public abstract class AbstractBlockTransactionSelectorTest {
                 txNotSelected, TransactionSelectionResult.PRIORITY_FEE_PER_GAS_BELOW_CURRENT_MIN));
   }
 
-  protected BlockTransactionSelector createBlockSelector(
+  @ParameterizedTest
+  @MethodSource("subsetOfPendingTransactionsIncludedWhenTxSelectionMaxTimeIsOver")
+  public void subsetOfPendingTransactionsIncludedWhenTxSelectionMaxTimeIsOver(
+      final boolean isPoa,
+      final boolean preProcessingTooLate,
+      final boolean processingTooLate,
+      final boolean postProcessingTooLate) {
+
+    final Supplier<Answer<TransactionSelectionResult>> inTime =
+        () -> invocation -> TransactionSelectionResult.SELECTED;
+    final BiFunction<Transaction, Long, Answer<TransactionSelectionResult>> tooLate =
+        (p, t) ->
+            invocation -> {
+              if (((PendingTransaction) invocation.getArgument(0)).getTransaction().equals(p)) {
+                Thread.sleep(t);
+              }
+              return TransactionSelectionResult.SELECTED;
+            };
+
+    final ProcessableBlockHeader blockHeader = createBlock(301_000);
+    final Address miningBeneficiary = AddressHelpers.ofValue(1);
+    final int poaMinBlockTime = 1;
+    final long blockTxsSelectionMaxTime = 750;
+    final long longProcessingTxTime = 500;
+
+    final List<Transaction> transactionsToInject = new ArrayList<>(3);
+    for (int i = 0; i < 2; i++) {
+      final Transaction tx = createTransaction(i, Wei.of(7), 100_000);
+      transactionsToInject.add(tx);
+      ensureTransactionIsValid(tx);
+    }
+
+    final Transaction lateTx = createTransaction(2, Wei.of(7), 100_000);
+    transactionsToInject.add(lateTx);
+    ensureTransactionIsValid(
+        lateTx, 0, 0, processingTooLate ? blockTxsSelectionMaxTime + longProcessingTxTime : 0);
+
+    PluginTransactionSelector transactionSelector = mock(PluginTransactionSelector.class);
+    when(transactionSelector.evaluateTransactionPreProcessing(any()))
+        .thenAnswer(
+            preProcessingTooLate
+                ? inTime.get()
+                : tooLate.apply(lateTx, blockTxsSelectionMaxTime + longProcessingTxTime));
+
+    when(transactionSelector.evaluateTransactionPostProcessing(any(), any()))
+        .thenAnswer(
+            postProcessingTooLate
+                ? inTime.get()
+                : tooLate.apply(lateTx, blockTxsSelectionMaxTime + longProcessingTxTime));
+
+    final PluginTransactionSelectorFactory transactionSelectorFactory =
+        mock(PluginTransactionSelectorFactory.class);
+    when(transactionSelectorFactory.create()).thenReturn(transactionSelector);
+
+    final BlockTransactionSelector selector =
+        createBlockSelectorAndSetupTxPool(
+            isPoa
+                ? createMiningParameters(
+                    Wei.ZERO, MIN_OCCUPANCY_100_PERCENT, poaMinBlockTime, Percentage.fromInt(75))
+                : createMiningParameters(
+                    Wei.ZERO, MIN_OCCUPANCY_100_PERCENT, blockTxsSelectionMaxTime),
+            transactionProcessor,
+            blockHeader,
+            miningBeneficiary,
+            Wei.ZERO,
+            transactionSelectorFactory);
+
+    transactionPool.addRemoteTransactions(transactionsToInject);
+
+    final TransactionSelectionResults results = selector.buildTransactionListForBlock();
+
+    // third tx is not selected, even if it could fit in the block,
+    // since the selection time was over
+    assertThat(results.getSelectedTransactions().size()).isEqualTo(2);
+
+    assertThat(results.getSelectedTransactions().containsAll(transactionsToInject.subList(0, 2)))
+        .isTrue();
+
+    assertThat(results.getReceipts().size()).isEqualTo(2);
+    assertThat(results.getCumulativeGasUsed()).isEqualTo(200_000);
+
+    // Ensure receipts have the correct cumulative gas
+    assertThat(results.getReceipts().get(0).getCumulativeGasUsed()).isEqualTo(100_000);
+    assertThat(results.getReceipts().get(1).getCumulativeGasUsed()).isEqualTo(200_000);
+
+    // given enough time we can check the not selected tx
+    await().until(() -> !results.getNotSelectedTransactions().isEmpty());
+    assertThat(results.getNotSelectedTransactions())
+        .containsOnly(entry(lateTx, TransactionSelectionResult.BLOCK_SELECTION_TIMEOUT));
+  }
+
+  private static Stream<Arguments>
+      subsetOfPendingTransactionsIncludedWhenTxSelectionMaxTimeIsOver() {
+
+    return Stream.of(
+        Arguments.of(false, true, false, false),
+        Arguments.of(false, false, true, false),
+        Arguments.of(false, false, false, true),
+        Arguments.of(true, true, false, false),
+        Arguments.of(true, false, true, false),
+        Arguments.of(true, false, false, true));
+  }
+
+  protected BlockTransactionSelector createBlockSelectorAndSetupTxPool(
+      final MiningParameters miningParameters,
       final MainnetTransactionProcessor transactionProcessor,
       final ProcessableBlockHeader blockHeader,
-      final Wei minGasPrice,
       final Address miningBeneficiary,
       final Wei blobGasPrice,
-      final double minBlockOccupancyRatio) {
-    final BlockTransactionSelector selector =
-        new BlockTransactionSelector(
-            miningParameters
-                .setMinTransactionGasPrice(minGasPrice)
-                .setMinBlockOccupancyRatio(minBlockOccupancyRatio),
+      final PluginTransactionSelectorFactory transactionSelectorFactory) {
+
+    transactionPool = createTransactionPool(miningParameters);
+
+    return createBlockSelector(
+        miningParameters,
         transactionProcessor,
-            blockchain,
-            worldState,
-            transactionPool,
         blockHeader,
-            this::createReceipt,
-            this::isCancelled,
         miningBeneficiary,
         blobGasPrice,
-            getFeeMarket(),
-            new LondonGasCalculator(),
-            GasLimitCalculator.constant(),
-            AllAcceptingTransactionSelector.INSTANCE);
-
-    return selector;
+        transactionSelectorFactory);
   }
 
-  protected BlockTransactionSelector createBlockSelectorWithTxSelPlugin(
+  protected BlockTransactionSelector createBlockSelector(
+      final MiningParameters miningParameters,
       final MainnetTransactionProcessor transactionProcessor,
       final ProcessableBlockHeader blockHeader,
       final Address miningBeneficiary,
       final Wei blobGasPrice,
       final PluginTransactionSelectorFactory transactionSelectorFactory) {
+
     final BlockTransactionSelector selector =
         new BlockTransactionSelector(
             miningParameters,
@@ -902,7 +1019,8 @@ public abstract class AbstractBlockTransactionSelectorTest {
             getFeeMarket(),
             new LondonGasCalculator(),
             GasLimitCalculator.constant(),
-            transactionSelectorFactory.create());
+            transactionSelectorFactory.create(),
+            ethScheduler);
 
     return selector;
   }
@@ -965,15 +1083,28 @@ public abstract class AbstractBlockTransactionSelectorTest {
 
   protected void ensureTransactionIsValid(
       final Transaction tx, final long gasUsedByTransaction, final long gasRemaining) {
+    ensureTransactionIsValid(tx, gasUsedByTransaction, gasRemaining, 0);
+  }
+
+  protected void ensureTransactionIsValid(
+      final Transaction tx,
+      final long gasUsedByTransaction,
+      final long gasRemaining,
+      final long processingTime) {
     when(transactionProcessor.processTransaction(
             any(), any(), any(), eq(tx), any(), any(), any(), anyBoolean(), any(), any()))
-        .thenReturn(
-            TransactionProcessingResult.successful(
+        .thenAnswer(
+            invocation -> {
+              if (processingTime > 0) {
+                Thread.sleep(processingTime);
+              }
+              return TransactionProcessingResult.successful(
                   new ArrayList<>(),
                   gasUsedByTransaction,
                   gasRemaining,
                   Bytes.EMPTY,
-                ValidationResult.valid()));
+                  ValidationResult.valid());
+            });
   }
 
   protected void ensureTransactionIsInvalid(
@@ -986,4 +1117,35 @@ public abstract class AbstractBlockTransactionSelectorTest {
   private BlockHeader blockHeader(final long number) {
     return new BlockHeaderTestFixture().number(number).buildHeader();
   }
+
+  protected MiningParameters createMiningParameters(
+      final Wei minGasPrice, final double minBlockOccupancyRatio, final long txsSelectionMaxTime) {
+    return ImmutableMiningParameters.builder()
+        .mutableInitValues(
+            MutableInitValues.builder()
+                .minTransactionGasPrice(minGasPrice)
+                .minBlockOccupancyRatio(minBlockOccupancyRatio)
+                .build())
+        .unstable(Unstable.builder().nonPoaBlockTxsSelectionMaxTime(txsSelectionMaxTime).build())
+        .build();
+  }
+
+  protected MiningParameters createMiningParameters(
+      final Wei minGasPrice,
+      final double minBlockOccupancyRatio,
+      final int minBlockTime,
+      final Percentage minBlockTimePercentage) {
+    return ImmutableMiningParameters.builder()
+        .mutableInitValues(
+            MutableInitValues.builder()
+                .minTransactionGasPrice(minGasPrice)
+                .minBlockOccupancyRatio(minBlockOccupancyRatio)
+                .build())
+        .unstable(
+            Unstable.builder()
+                .minBlockTime(minBlockTime)
+                .poaBlockTxsSelectionMaxTime(minBlockTimePercentage)
+                .build())
+        .build();
+  }
 }

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/LegacyFeeMarketBlockTransactionSelectorTest.java

@@ -20,6 +20,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.config.GenesisConfigFile;
+import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
@@ -61,7 +62,7 @@ public class LegacyFeeMarketBlockTransactionSelectorTest
   }
 
   @Override
-  protected TransactionPool createTransactionPool() {
+  protected TransactionPool createTransactionPool(final MiningParameters miningParameters) {
     final TransactionPoolConfiguration poolConf =
         ImmutableTransactionPoolConfiguration.builder()
             .txPoolMaxSize(5)

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/LondonFeeMarketBlockTransactionSelectorTest.java

@@ -16,6 +16,7 @@ package org.hyperledger.besu.ethereum.blockcreation;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
+import static org.hyperledger.besu.ethereum.core.MiningParameters.Unstable.DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
 import static org.mockito.Mockito.mock;
 
 import org.hyperledger.besu.config.GenesisConfigFile;
@@ -24,6 +25,8 @@ import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.blockcreation.txselection.BlockTransactionSelector;
 import org.hyperledger.besu.ethereum.blockcreation.txselection.TransactionSelectionResults;
 import org.hyperledger.besu.ethereum.core.AddressHelpers;
+import org.hyperledger.besu.ethereum.core.ImmutableMiningParameters;
+import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.core.ProcessableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
@@ -71,7 +74,7 @@ public class LondonFeeMarketBlockTransactionSelectorTest
   }
 
   @Override
-  protected TransactionPool createTransactionPool() {
+  protected TransactionPool createTransactionPool(final MiningParameters miningParameters) {
     final TransactionPoolConfiguration poolConf =
         ImmutableTransactionPoolConfiguration.builder()
             .txPoolMaxSize(5)
@@ -105,13 +108,14 @@ public class LondonFeeMarketBlockTransactionSelectorTest
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.of(6), MIN_OCCUPANCY_80_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
-            Wei.of(6),
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     // tx is willing to pay max 7 wei for gas, but current network condition (baseFee == 1)
     // result in it paying 2 wei, that is below the minimum accepted by the node, so it is skipped
@@ -133,13 +137,14 @@ public class LondonFeeMarketBlockTransactionSelectorTest
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.of(6), MIN_OCCUPANCY_80_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
-            Wei.of(6),
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     // tx is willing to pay max 7 wei for gas, and current network condition (baseFee == 5)
     // result in it paying the max, that is >= the minimum accepted by the node, so it is selected
@@ -160,13 +165,14 @@ public class LondonFeeMarketBlockTransactionSelectorTest
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            createMiningParameters(
+                Wei.of(6), MIN_OCCUPANCY_80_PERCENT, DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME),
             transactionProcessor,
             blockHeader,
-            Wei.of(6),
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
 
     // tx is willing to pay max 7 wei for gas, but current network condition (baseFee == 1)
     // result in it paying 2 wei, that is below the minimum accepted by the node, but since it is
@@ -192,8 +198,6 @@ public class LondonFeeMarketBlockTransactionSelectorTest
     final Transaction txFrontier2 = createTransaction(2, Wei.of(7L), 100_000);
     final Transaction txLondon2 = createEIP1559Transaction(3, Wei.ONE, Wei.ONE, 100_000);
 
-    transactionPool.addRemoteTransactions(List.of(txFrontier1, txLondon1, txFrontier2, txLondon2));
-
     ensureTransactionIsValid(txFrontier1);
     ensureTransactionIsValid(txLondon1);
     ensureTransactionIsValid(txFrontier2);
@@ -201,13 +205,15 @@ public class LondonFeeMarketBlockTransactionSelectorTest
 
     final Address miningBeneficiary = AddressHelpers.ofValue(1);
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            defaultTestMiningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             miningBeneficiary,
             Wei.ZERO,
-            MIN_OCCUPANCY_80_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    transactionPool.addRemoteTransactions(List.of(txFrontier1, txLondon1, txFrontier2, txLondon2));
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 
@@ -220,6 +226,8 @@ public class LondonFeeMarketBlockTransactionSelectorTest
   @Override
   public void shouldNotSelectTransactionsWithPriorityFeeLessThanConfig() {
     ProcessableBlockHeader blockHeader = createBlock(5_000_000, Wei.ONE);
+    final MiningParameters miningParameters =
+        ImmutableMiningParameters.builder().from(defaultTestMiningParameters).build();
     miningParameters.setMinPriorityFeePerGas(Wei.of(7));
 
     final Transaction txSelected1 = createEIP1559Transaction(1, Wei.of(8), Wei.of(8), 100_000);
@@ -237,19 +245,19 @@ public class LondonFeeMarketBlockTransactionSelectorTest
     final Transaction txNotSelected2 = createEIP1559Transaction(4, Wei.of(8), Wei.of(6), 100_000);
     ensureTransactionIsValid(txNotSelected2);
 
-    transactionPool.addRemoteTransactions(
-        List.of(txSelected1, txNotSelected1, txSelected2, txNotSelected2));
-
-    assertThat(transactionPool.getPendingTransactions().size()).isEqualTo(4);
-
     final BlockTransactionSelector selector =
-        createBlockSelector(
+        createBlockSelectorAndSetupTxPool(
+            miningParameters,
             transactionProcessor,
             blockHeader,
-            Wei.ZERO,
             AddressHelpers.ofValue(1),
             Wei.ZERO,
-            MIN_OCCUPANCY_100_PERCENT);
+            NO_PLUGIN_TRANSACTION_SELECTOR_FACTORY);
+
+    transactionPool.addRemoteTransactions(
+        List.of(txSelected1, txNotSelected1, txSelected2, txNotSelected2));
+
+    assertThat(transactionPool.getPendingTransactions().size()).isEqualTo(4);
 
     final TransactionSelectionResults results = selector.buildTransactionListForBlock();
 

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/PoWBlockCreatorTest.java

@@ -115,7 +115,8 @@ class PoWBlockCreatorTest extends AbstractBlockCreatorTest {
             executionContextTestFixture.getProtocolContext(),
             executionContextTestFixture.getProtocolSchedule(),
             solver,
-            executionContextTestFixture.getBlockchain().getChainHeadHeader());
+            executionContextTestFixture.getBlockchain().getChainHeadHeader(),
+            ethScheduler);
 
     // A Hashrate should not exist in the block creator prior to creating a block
     assertThat(blockCreator.getHashesPerSecond()).isNotPresent();
@@ -168,7 +169,8 @@ class PoWBlockCreatorTest extends AbstractBlockCreatorTest {
             executionContextTestFixture.getProtocolContext(),
             executionContextTestFixture.getProtocolSchedule(),
             solver,
-            executionContextTestFixture.getBlockchain().getChainHeadHeader());
+            executionContextTestFixture.getBlockchain().getChainHeadHeader(),
+            ethScheduler);
 
     assertThat(blockCreator.createBlock(BLOCK_1_TIMESTAMP)).isNotNull();
     // If we weren't setting difficulty to 2^256-1 a difficulty of 1 would have caused a
@@ -212,7 +214,8 @@ class PoWBlockCreatorTest extends AbstractBlockCreatorTest {
             executionContextTestFixture.getProtocolContext(),
             executionContextTestFixture.getProtocolSchedule(),
             solver,
-            executionContextTestFixture.getBlockchain().getChainHeadHeader());
+            executionContextTestFixture.getBlockchain().getChainHeadHeader(),
+            ethScheduler);
 
     final MutableWorldState mutableWorldState =
         executionContextTestFixture.getStateArchive().getMutable();
@@ -278,7 +281,8 @@ class PoWBlockCreatorTest extends AbstractBlockCreatorTest {
             executionContextTestFixture.getProtocolContext(),
             executionContextTestFixture.getProtocolSchedule(),
             solver,
-            executionContextTestFixture.getBlockchain().getChainHeadHeader());
+            executionContextTestFixture.getBlockchain().getChainHeadHeader(),
+            ethScheduler);
 
     final MutableWorldState mutableWorldState =
         executionContextTestFixture.getStateArchive().getMutable();

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/PoWMinerExecutorTest.java

@@ -24,6 +24,7 @@ import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -34,6 +35,7 @@ import org.hyperledger.besu.ethereum.mainnet.EpochCalculator;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 import org.hyperledger.besu.util.Subscribers;
 
@@ -44,6 +46,7 @@ import org.junit.jupiter.api.Test;
 
 public class PoWMinerExecutorTest {
   private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
+  private final EthScheduler ethScheduler = new DeterministicEthScheduler();
 
   @Test
   public void startingMiningWithoutCoinbaseThrowsException() {
@@ -58,7 +61,8 @@ public class PoWMinerExecutorTest {
             transactionPool,
             miningParameters,
             new DefaultBlockScheduler(1, 10, TestClock.fixed()),
-            new EpochCalculator.DefaultEpochCalculator());
+            new EpochCalculator.DefaultEpochCalculator(),
+            ethScheduler);
 
     assertThatExceptionOfType(CoinbaseNotSetException.class)
         .isThrownBy(() -> executor.startAsyncMining(Subscribers.create(), Subscribers.none(), null))
@@ -78,7 +82,8 @@ public class PoWMinerExecutorTest {
             transactionPool,
             miningParameters,
             new DefaultBlockScheduler(1, 10, TestClock.fixed()),
-            new EpochCalculator.DefaultEpochCalculator());
+            new EpochCalculator.DefaultEpochCalculator(),
+            ethScheduler);
 
     assertThatExceptionOfType(IllegalArgumentException.class)
         .isThrownBy(() -> executor.setCoinbase(null))

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/MiningParameters.java

@@ -16,12 +16,16 @@ package org.hyperledger.besu.ethereum.core;
 
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Wei;
+import org.hyperledger.besu.util.number.Percentage;
 
 import java.time.Duration;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalInt;
 import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.tuweni.bytes.Bytes;
 import org.immutables.value.Value;
 
@@ -34,6 +38,7 @@ public abstract class MiningParameters {
               ImmutableMiningParameters.MutableInitValues.builder().isMiningEnabled(false).build())
           .build();
 
+  @VisibleForTesting
   public static final MiningParameters newDefault() {
     return ImmutableMiningParameters.builder().build();
   }
@@ -261,6 +266,8 @@ public abstract class MiningParameters {
     int DEFAULT_MAX_OMMERS_DEPTH = 8;
     long DEFAULT_POS_BLOCK_CREATION_MAX_TIME = Duration.ofSeconds(12).toMillis();
     long DEFAULT_POS_BLOCK_CREATION_REPETITION_MIN_DURATION = Duration.ofMillis(500).toMillis();
+    long DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME = Duration.ofSeconds(5).toMillis();
+    Percentage DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME = Percentage.fromInt(75);
 
     MiningParameters.Unstable DEFAULT = ImmutableMiningParameters.Unstable.builder().build();
 
@@ -298,5 +305,27 @@ public abstract class MiningParameters {
     default String getStratumExtranonce() {
       return "080c";
     }
+
+    @Value.Default
+    default long getNonPoaBlockTxsSelectionMaxTime() {
+      return DEFAULT_NON_POA_BLOCK_TXS_SELECTION_MAX_TIME;
+    }
+
+    @Value.Default
+    default Percentage getPoaBlockTxsSelectionMaxTime() {
+      return DEFAULT_POA_BLOCK_TXS_SELECTION_MAX_TIME;
+    }
+
+    OptionalInt getMinBlockTime();
+
+    @Value.Derived
+    default long getBlockTxsSelectionMaxTime() {
+      if (getMinBlockTime().isPresent()) {
+        return (TimeUnit.SECONDS.toMillis(getMinBlockTime().getAsInt())
+                * getPoaBlockTxsSelectionMaxTime().getValue())
+            / 100;
+      }
+      return getNonPoaBlockTxsSelectionMaxTime();
+    }
   }
 }

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/bonsai/AbstractIsolationTests.java

@@ -49,6 +49,7 @@ import org.hyperledger.besu.ethereum.core.SealableBlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.TransactionTestFixture;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
@@ -72,6 +73,7 @@ import org.hyperledger.besu.plugin.services.BesuConfiguration;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDBKeyValueStorageFactory;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDBMetricsFactory;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBFactoryConfiguration;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.nio.file.Path;
 import java.util.Arrays;
@@ -92,6 +94,7 @@ public abstract class AbstractIsolationTests {
   protected BonsaiWorldStateKeyValueStorage bonsaiWorldStateStorage;
   protected ProtocolContext protocolContext;
   protected EthContext ethContext;
+  protected EthScheduler ethScheduler = new DeterministicEthScheduler();
   final Function<String, KeyPair> asKeyPair =
       key ->
           SignatureAlgorithmFactory.getInstance()
@@ -214,7 +217,8 @@ public abstract class AbstractIsolationTests {
         final TransactionPool transactionPool,
         final ProtocolContext protocolContext,
         final ProtocolSchedule protocolSchedule,
-        final BlockHeader parentHeader) {
+        final BlockHeader parentHeader,
+        final EthScheduler ethScheduler) {
       super(
           miningParameters,
           miningBeneficiaryCalculator,
@@ -223,14 +227,16 @@ public abstract class AbstractIsolationTests {
           protocolContext,
           protocolSchedule,
           parentHeader,
-          Optional.empty());
+          Optional.empty(),
+          ethScheduler);
     }
 
     static TestBlockCreator forHeader(
         final BlockHeader parentHeader,
         final ProtocolContext protocolContext,
         final ProtocolSchedule protocolSchedule,
-        final TransactionPool transactionPool) {
+        final TransactionPool transactionPool,
+        final EthScheduler ethScheduler) {
 
       final MiningParameters miningParameters =
           ImmutableMiningParameters.builder()
@@ -251,7 +257,8 @@ public abstract class AbstractIsolationTests {
           transactionPool,
           protocolContext,
           protocolSchedule,
-          parentHeader);
+          parentHeader,
+          ethScheduler);
     }
 
     @Override
@@ -282,7 +289,8 @@ public abstract class AbstractIsolationTests {
 
   protected Block forTransactions(
       final List<Transaction> transactions, final BlockHeader forHeader) {
-    return TestBlockCreator.forHeader(forHeader, protocolContext, protocolSchedule, transactionPool)
+    return TestBlockCreator.forHeader(
+            forHeader, protocolContext, protocolSchedule, transactionPool, ethScheduler)
         .createBlock(transactions, Collections.emptyList(), System.currentTimeMillis())
         .getBlock();
   }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthScheduler.java

@@ -53,6 +53,7 @@ public class EthScheduler {
   protected final ExecutorService txWorkerExecutor;
   protected final ExecutorService servicesExecutor;
   protected final ExecutorService computationExecutor;
+  protected final ExecutorService blockCreationExecutor;
 
   private final Collection<CompletableFuture<?>> pendingFutures = new ConcurrentLinkedDeque<>();
 
@@ -87,7 +88,9 @@ public class EthScheduler {
             EthScheduler.class.getSimpleName() + "-Computation",
             1,
             computationWorkerCount,
-            metricsSystem));
+            metricsSystem),
+        MonitoredExecutors.newCachedThreadPool(
+            EthScheduler.class.getSimpleName() + "-BlockCreation", metricsSystem));
   }
 
   protected EthScheduler(
@@ -95,12 +98,14 @@ public class EthScheduler {
       final ScheduledExecutorService scheduler,
       final ExecutorService txWorkerExecutor,
       final ExecutorService servicesExecutor,
-      final ExecutorService computationExecutor) {
+      final ExecutorService computationExecutor,
+      final ExecutorService blockCreationExecutor) {
     this.syncWorkerExecutor = syncWorkerExecutor;
     this.scheduler = scheduler;
     this.txWorkerExecutor = txWorkerExecutor;
     this.servicesExecutor = servicesExecutor;
     this.computationExecutor = computationExecutor;
+    this.blockCreationExecutor = blockCreationExecutor;
   }
 
   public <T> CompletableFuture<T> scheduleSyncWorkerTask(
@@ -202,6 +207,10 @@ public class EthScheduler {
     return promise;
   }
 
+  public CompletableFuture<Void> scheduleBlockCreationTask(final Runnable task) {
+    return CompletableFuture.runAsync(task, blockCreationExecutor);
+  }
+
   public <T> CompletableFuture<T> timeout(final EthTask<T> task) {
     return timeout(task, defaultTimeout);
   }

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTest.java

@@ -1088,8 +1088,9 @@ public final class EthProtocolManagerTest {
     final ExecutorService transactions = mock(ExecutorService.class);
     final ExecutorService services = mock(ExecutorService.class);
     final ExecutorService computations = mock(ExecutorService.class);
+    final ExecutorService blockCreation = mock(ExecutorService.class);
     final EthScheduler ethScheduler =
-        new EthScheduler(worker, scheduled, transactions, services, computations);
+        new EthScheduler(worker, scheduled, transactions, services, computations, blockCreation);
 
     // Create the fake TransactionMessage to feed to the EthManager.
     final BlockDataGenerator gen = new BlockDataGenerator(1);

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTestUtil.java

@@ -27,7 +27,6 @@ import org.hyperledger.besu.ethereum.core.Difficulty;
 import org.hyperledger.besu.ethereum.core.ProtocolScheduleFixture;
 import org.hyperledger.besu.ethereum.eth.EthProtocol;
 import org.hyperledger.besu.ethereum.eth.EthProtocolConfiguration;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler.TimeoutPolicy;
 import org.hyperledger.besu.ethereum.eth.manager.snap.SnapProtocolManager;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
@@ -39,6 +38,8 @@ import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageFormat;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler.TimeoutPolicy;
 import org.hyperledger.besu.testutil.TestClock;
 
 import java.math.BigInteger;

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthSchedulerShutdownTest.java

@@ -32,6 +32,7 @@ public class EthSchedulerShutdownTest {
   private ExecutorService txWorkerExecutor;
   private ExecutorService servicesExecutor;
   private ExecutorService computationExecutor;
+  private ExecutorService blockCreationExecutor;
 
   @BeforeEach
   public void setup() {
@@ -40,13 +41,15 @@ public class EthSchedulerShutdownTest {
     txWorkerExecutor = Executors.newSingleThreadExecutor();
     servicesExecutor = Executors.newSingleThreadExecutor();
     computationExecutor = Executors.newSingleThreadExecutor();
+    blockCreationExecutor = Executors.newSingleThreadExecutor();
     ethScheduler =
         new EthScheduler(
             syncWorkerExecutor,
             scheduledExecutor,
             txWorkerExecutor,
             servicesExecutor,
-            computationExecutor);
+            computationExecutor,
+            blockCreationExecutor);
   }
 
   @Test

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthSchedulerTest.java

@@ -21,7 +21,9 @@ import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.MockExecutorService;
+import org.hyperledger.besu.testutil.MockScheduledExecutor;
 
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java

@@ -28,7 +28,6 @@ import org.hyperledger.besu.ethereum.core.BlockchainSetupUtil;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.eth.EthProtocol;
 import org.hyperledger.besu.ethereum.eth.EthProtocolConfiguration;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthMessages;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
@@ -46,6 +45,7 @@ import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageFormat;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.TestClock;
 
 import java.time.ZoneId;

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/peervalidation/AbstractPeerBlockValidatorTest.java

@@ -22,7 +22,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
 import org.hyperledger.besu.ethereum.core.BlockDataGenerator.BlockOptions;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManagerTestUtil;
@@ -30,6 +29,7 @@ import org.hyperledger.besu.ethereum.eth.manager.RespondingEthPeer;
 import org.hyperledger.besu.ethereum.eth.messages.BlockHeadersMessage;
 import org.hyperledger.besu.ethereum.eth.messages.EthPV62;
 import org.hyperledger.besu.ethereum.eth.messages.GetBlockHeadersMessage;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardSyncStepTest.java

@@ -29,7 +29,6 @@ import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManagerTestUtil;
@@ -39,6 +38,7 @@ import org.hyperledger.besu.ethereum.mainnet.MainnetBlockHeaderFunctions;
 import org.hyperledger.besu.ethereum.mainnet.MainnetProtocolSchedule;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.services.kvstore.InMemoryKeyValueStorage;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.nio.charset.StandardCharsets;
 import java.util.List;

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/FastWorldStateDownloaderTest.java

@@ -34,7 +34,6 @@ import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
 import org.hyperledger.besu.ethereum.core.MutableWorldState;
 import org.hyperledger.besu.ethereum.core.ProtocolScheduleFixture;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManagerTestUtil;
@@ -68,6 +67,7 @@ import org.hyperledger.besu.evm.worldstate.WorldState;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.services.kvstore.InMemoryKeyValueStorage;
 import org.hyperledger.besu.services.tasks.InMemoryTasksPriorityQueues;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 import org.hyperledger.besu.testutil.MockExecutorService;
 import org.hyperledger.besu.testutil.TestClock;
 

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/DynamicPivotBlockManagerTest.java

@@ -22,11 +22,11 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
-import org.hyperledger.besu.ethereum.eth.manager.DeterministicEthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.sync.fastsync.FastSyncActions;
 import org.hyperledger.besu.ethereum.eth.sync.fastsync.FastSyncState;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
 
 import java.util.OptionalLong;
 

ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/RetestethContext.java

@@ -290,6 +290,14 @@ public class RetestethContext {
     return protocolContext;
   }
 
+  public EthScheduler getEthScheduler() {
+    return ethScheduler;
+  }
+
+  public void setEthScheduler(final EthScheduler ethScheduler) {
+    this.ethScheduler = ethScheduler;
+  }
+
   public long getBlockHeight() {
     return blockchain.getChainHeadBlockNumber();
   }

ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/methods/TestMineBlocks.java

@@ -69,7 +69,8 @@ public class TestMineBlocks implements JsonRpcMethod {
             protocolContext,
             protocolSchedule,
             context.getEthHashSolver(),
-            blockchain.getChainHeadHeader());
+            blockchain.getChainHeadHeader(),
+            context.getEthScheduler());
     final Block block =
         blockCreator.createBlock(retesethClock.instant().getEpochSecond()).getBlock();
 

plugin-api/build.gradle

@@ -69,7 +69,7 @@ Calculated : ${currentHash}
 tasks.register('checkAPIChanges', FileStateChecker) {
   description = "Checks that the API for the Plugin-API project does not change without deliberate thought"
   files = sourceSets.main.allJava.files
-  knownHash = 'n4WfeMvltN4XWGtztd8ABjSU2TLiI3tk5yABivkgsFA='
+  knownHash = '7Aj0APsKs1wBVqaWQFdEs85/MNKxTiVzyjIeZ+zCWlw='
 }
 check.dependsOn('checkAPIChanges')
 

plugin-api/src/main/java/org/hyperledger/besu/plugin/data/TransactionSelectionResult.java

@@ -28,6 +28,7 @@ public class TransactionSelectionResult {
     SELECTED,
     BLOCK_FULL(true, false),
     BLOCK_OCCUPANCY_ABOVE_THRESHOLD(true, false),
+    BLOCK_SELECTION_TIMEOUT(true, false),
     INVALID_TRANSIENT(false, false),
     INVALID(false, true);
 
@@ -56,6 +57,11 @@ public class TransactionSelectionResult {
   /** The transaction has not been selected since the block is full. */
   public static final TransactionSelectionResult BLOCK_FULL =
       new TransactionSelectionResult(Status.BLOCK_FULL);
+  /** There was no more time to add transaction to the block */
+  public static final TransactionSelectionResult BLOCK_SELECTION_TIMEOUT =
+      new TransactionSelectionResult(Status.BLOCK_SELECTION_TIMEOUT);
+  ;
+
   /**
    * The transaction has not been selected since too large and the occupancy of the block is enough
    * to stop the selection.

testutil/build.gradle

@@ -28,6 +28,7 @@ jar {
 }
 
 dependencies {
+  implementation project(':ethereum:eth')
   implementation project(':plugin-api')
   implementation project(':util')
 

testutil/src/main/java/org/hyperledger/besu/testutil/DeterministicEthScheduler.java

@@ -12,9 +12,9 @@
  *
  * SPDX-License-Identifier: Apache-2.0
  */
-package org.hyperledger.besu.ethereum.eth.manager;
+package org.hyperledger.besu.testutil;
 
-import org.hyperledger.besu.testutil.MockExecutorService;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -32,16 +32,23 @@ public class DeterministicEthScheduler extends EthScheduler {
   private final List<MockExecutorService> executors;
   private final List<PendingTimeout<?>> pendingTimeouts = new ArrayList<>();
 
+  /** Create a new deterministic scheduler that never timeouts */
   public DeterministicEthScheduler() {
     this(TimeoutPolicy.NEVER_TIMEOUT);
   }
 
+  /**
+   * Create a new deterministic scheduler with the provided timeout policy
+   *
+   * @param timeoutPolicy the timeout policy
+   */
   public DeterministicEthScheduler(final TimeoutPolicy timeoutPolicy) {
     super(
         new MockExecutorService(),
         new MockScheduledExecutor(),
         new MockExecutorService(),
         new MockExecutorService(),
+        new MockExecutorService(),
         new MockExecutorService());
 
     this.timeoutPolicy = timeoutPolicy;
@@ -51,40 +58,72 @@ public class DeterministicEthScheduler extends EthScheduler {
             (MockExecutorService) this.scheduler,
             (MockExecutorService) this.txWorkerExecutor,
             (MockExecutorService) this.servicesExecutor,
-            (MockExecutorService) this.computationExecutor);
+            (MockExecutorService) this.computationExecutor,
+            (MockExecutorService) this.blockCreationExecutor);
   }
 
-  // Test utility for running pending futures
+  /** Test utility for manually running pending futures, when autorun is disabled */
   public void runPendingFutures() {
     executors.forEach(MockExecutorService::runPendingFutures);
   }
 
+  /**
+   * Get the count of pending tasks
+   *
+   * @return the count of pending tasks
+   */
   public long getPendingFuturesCount() {
     return executors.stream().mapToLong(MockExecutorService::getPendingFuturesCount).sum();
   }
 
+  /** Expire all pending timeouts */
   public void expirePendingTimeouts() {
     final List<PendingTimeout<?>> toExpire = new ArrayList<>(pendingTimeouts);
     pendingTimeouts.clear();
     toExpire.forEach(PendingTimeout::expire);
   }
 
+  /** Do not automatically run submitted tasks. Tasks can be later run using runPendingFutures */
   public void disableAutoRun() {
     executors.forEach(e -> e.setAutoRun(false));
   }
 
-  MockExecutorService mockSyncWorkerExecutor() {
+  /**
+   * Get the sync worker mock executor
+   *
+   * @return the mock executor
+   */
+  public MockExecutorService mockSyncWorkerExecutor() {
     return (MockExecutorService) syncWorkerExecutor;
   }
 
-  MockScheduledExecutor mockScheduledExecutor() {
+  /**
+   * Get the scheduled mock executor
+   *
+   * @return the mock executor
+   */
+  public MockScheduledExecutor mockScheduledExecutor() {
     return (MockScheduledExecutor) scheduler;
   }
 
+  /**
+   * Get the service mock executor
+   *
+   * @return the mock executor
+   */
   public MockExecutorService mockServiceExecutor() {
     return (MockExecutorService) servicesExecutor;
   }
 
+  /**
+   * Get the block creation mock executor
+   *
+   * @return the mock executor
+   */
+  public MockExecutorService mockBlockCreationExecutor() {
+    return (MockExecutorService) blockCreationExecutor;
+  }
+
   @Override
   public <T> void failAfterTimeout(final CompletableFuture<T> promise, final Duration timeout) {
     final PendingTimeout<T> pendingTimeout = new PendingTimeout<>(promise, timeout);
@@ -95,13 +134,27 @@ public class DeterministicEthScheduler extends EthScheduler {
     }
   }
 
+  /** Used to define the timeout behavior of the scheduler */
   @FunctionalInterface
   public interface TimeoutPolicy {
+    /** A policy that never timeouts */
     TimeoutPolicy NEVER_TIMEOUT = () -> false;
+    /** A policy that timeouts on every task */
     TimeoutPolicy ALWAYS_TIMEOUT = () -> true;
 
+    /**
+     * If it should simulate a timeout when called
+     *
+     * @return true if the scheduler should timeouts
+     */
     boolean shouldTimeout();
 
+    /**
+     * Create a timeout policy that timeouts x times
+     *
+     * @param times the number of timeouts
+     * @return the timeout policy
+     */
     static TimeoutPolicy timeoutXTimes(final int times) {
       final AtomicInteger timeouts = new AtomicInteger(times);
       return () -> {

testutil/src/main/java/org/hyperledger/besu/testutil/MockScheduledExecutor.java

@@ -12,9 +12,7 @@
  *
  * SPDX-License-Identifier: Apache-2.0
  */
-package org.hyperledger.besu.ethereum.eth.manager;
-
-import org.hyperledger.besu.testutil.MockExecutorService;
+package org.hyperledger.besu.testutil;
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.Delayed;
@@ -25,6 +23,7 @@ import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+/** The mock scheduled executor */
 public class MockScheduledExecutor extends MockExecutorService implements ScheduledExecutorService {
 
   @Override