Feature/fleet mode rebase (#6641)

* fleet mode squash commit rebase

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

acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java

@@ -180,11 +180,10 @@ public class ThreadBesuNodeRunner implements BesuNodeRunner {
             .from(node.getMiningParameters())
             .transactionSelectionService(transactionSelectionServiceImpl)
             .build();
-    commonPluginConfiguration.init(
+    commonPluginConfiguration
-        dataDir,
+        .init(dataDir, dataDir.resolve(DATABASE_PATH), node.getDataStorageConfiguration())
-        dataDir.resolve(DATABASE_PATH),
+        .withMiningParameters(miningParameters);
-        node.getDataStorageConfiguration(),
+
-        miningParameters);
     final BesuPluginContextImpl besuPluginContext =
         besuPluginContextMap.computeIfAbsent(
             node,

acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/privacy/PrivacyNode.java

@@ -274,7 +274,9 @@ public class PrivacyNode implements AutoCloseable {
   private PrivacyStorageProvider createKeyValueStorageProvider(
       final Path dataLocation, final Path dbLocation) {
     final var besuConfiguration = new BesuConfigurationImpl();
-    besuConfiguration.init(dataLocation, dbLocation, null, besuConfig.getMiningParameters());
+    besuConfiguration
+        .init(dataLocation, dbLocation, null)
+        .withMiningParameters(besuConfig.getMiningParameters());
     return new PrivacyKeyValueStorageProviderBuilder()
         .withStorageFactory(
             new RocksDBKeyValuePrivacyStorageFactory(

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

@@ -25,6 +25,7 @@ import org.hyperledger.besu.ethereum.api.query.cache.TransactionLogBloomCacher;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolEvictionService;
 import org.hyperledger.besu.ethereum.p2p.network.NetworkRunner;
+import org.hyperledger.besu.ethereum.p2p.network.P2PNetwork;
 import org.hyperledger.besu.ethereum.stratum.StratumServer;
 import org.hyperledger.besu.ethstats.EthStatsService;
 import org.hyperledger.besu.metrics.MetricsService;
@@ -422,6 +423,15 @@ public class Runner implements AutoCloseable {
     return networkRunner.getNetwork().getLocalEnode();
   }
 
+  /**
+   * get P2PNetwork service.
+   *
+   * @return p2p network service.
+   */
+  public P2PNetwork getP2PNetwork() {
+    return networkRunner.getNetwork();
+  }
+
   @FunctionalInterface
   private interface SynchronousShutdown {
     /**

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

@@ -172,21 +172,29 @@ import org.hyperledger.besu.plugin.services.TransactionSimulationService;
 import org.hyperledger.besu.plugin.services.exception.StorageException;
 import org.hyperledger.besu.plugin.services.metrics.MetricCategory;
 import org.hyperledger.besu.plugin.services.metrics.MetricCategoryRegistry;
+import org.hyperledger.besu.plugin.services.p2p.P2PService;
+import org.hyperledger.besu.plugin.services.rlp.RlpConverterService;
 import org.hyperledger.besu.plugin.services.securitymodule.SecurityModule;
 import org.hyperledger.besu.plugin.services.storage.DataStorageFormat;
 import org.hyperledger.besu.plugin.services.storage.PrivacyKeyValueStorageFactory;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDBPlugin;
+import org.hyperledger.besu.plugin.services.sync.SynchronizationService;
+import org.hyperledger.besu.plugin.services.transactionpool.TransactionPoolService;
 import org.hyperledger.besu.services.BesuConfigurationImpl;
 import org.hyperledger.besu.services.BesuEventsImpl;
 import org.hyperledger.besu.services.BesuPluginContextImpl;
 import org.hyperledger.besu.services.BlockchainServiceImpl;
+import org.hyperledger.besu.services.P2PServiceImpl;
 import org.hyperledger.besu.services.PermissioningServiceImpl;
 import org.hyperledger.besu.services.PicoCLIOptionsImpl;
 import org.hyperledger.besu.services.PrivacyPluginServiceImpl;
+import org.hyperledger.besu.services.RlpConverterServiceImpl;
 import org.hyperledger.besu.services.RpcEndpointServiceImpl;
 import org.hyperledger.besu.services.SecurityModuleServiceImpl;
 import org.hyperledger.besu.services.StorageServiceImpl;
+import org.hyperledger.besu.services.SynchronizationServiceImpl;
 import org.hyperledger.besu.services.TraceServiceImpl;
+import org.hyperledger.besu.services.TransactionPoolServiceImpl;
 import org.hyperledger.besu.services.TransactionPoolValidatorServiceImpl;
 import org.hyperledger.besu.services.TransactionSelectionServiceImpl;
 import org.hyperledger.besu.services.TransactionSimulationServiceImpl;
@@ -1161,12 +1169,15 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
       final var runner = buildRunner();
       runner.startExternalServices();
 
-      startPlugins();
+      startPlugins(runner);
       validatePluginOptions();
       setReleaseMetrics();
       preSynchronization();
 
       runner.startEthereumMainLoop();
+
+      besuPluginContext.afterExternalServicesMainLoop();
+
       runner.awaitStop();
 
     } catch (final Exception e) {
@@ -1327,7 +1338,7 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
         pidPath);
   }
 
-  private void startPlugins() {
+  private void startPlugins(final Runner runner) {
     blockchainServiceImpl.init(
         besuController.getProtocolContext(), besuController.getProtocolSchedule());
     transactionSimulationServiceImpl.init(
@@ -1348,6 +1359,26 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
             besuController.getProtocolContext().getBadBlockManager()));
     besuPluginContext.addService(MetricsSystem.class, getMetricsSystem());
 
+    besuPluginContext.addService(BlockchainService.class, blockchainServiceImpl);
+
+    besuPluginContext.addService(
+        SynchronizationService.class,
+        new SynchronizationServiceImpl(
+            besuController.getProtocolContext(),
+            besuController.getProtocolSchedule(),
+            besuController.getSyncState(),
+            besuController.getProtocolContext().getWorldStateArchive()));
+
+    besuPluginContext.addService(P2PService.class, new P2PServiceImpl(runner.getP2PNetwork()));
+
+    besuPluginContext.addService(
+        TransactionPoolService.class,
+        new TransactionPoolServiceImpl(besuController.getTransactionPool()));
+
+    besuPluginContext.addService(
+        RlpConverterService.class,
+        new RlpConverterServiceImpl(besuController.getProtocolSchedule()));
+
     besuPluginContext.addService(
         TraceService.class,
         new TraceServiceImpl(
@@ -1653,11 +1684,11 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
   private void validateChainDataPruningParams() {
     if (unstableChainPruningOptions.getChainDataPruningEnabled()
         && unstableChainPruningOptions.getChainDataPruningBlocksRetained()
-            < ChainPruningOptions.DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED) {
+            < unstableChainPruningOptions.getChainDataPruningBlocksRetainedLimit()) {
       throw new ParameterException(
           this.commandLine,
           "--Xchain-pruning-blocks-retained must be >= "
-              + ChainPruningOptions.DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED);
+              + unstableChainPruningOptions.getChainDataPruningBlocksRetainedLimit());
     }
   }
 
@@ -1843,11 +1874,10 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
    * @return instance of BesuControllerBuilder
    */
   public BesuControllerBuilder getControllerBuilder() {
-    pluginCommonConfiguration.init(
+    pluginCommonConfiguration
-        dataDir(),
+        .init(dataDir(), dataDir().resolve(DATABASE_PATH), getDataStorageConfiguration())
-        dataDir().resolve(DATABASE_PATH),
+        .withMiningParameters(getMiningParameters())
-        getDataStorageConfiguration(),
+        .withJsonRpcHttpOptions(jsonRpcHttpOptions);
-        miningParametersSupplier.get());
     final KeyValueStorageProvider storageProvider = keyValueStorageProvider(keyValueStorageName);
     return controllerBuilderFactory
         .fromEthNetworkConfig(updateNetworkConfig(network), getDefaultSyncModeIfNotSet())

besu/src/main/java/org/hyperledger/besu/cli/options/stable/JsonRpcHttpOptions.java

@@ -475,6 +475,15 @@ public class JsonRpcHttpOptions {
     return rpcHttpApis;
   }
 
+  /**
+   * Returns the host for RPC over HTTP.
+   *
+   * @return The port number
+   */
+  public String getRpcHttpHost() {
+    return rpcHttpHost;
+  }
+
   /**
    * Returns the port for RPC over HTTP.
    *

besu/src/main/java/org/hyperledger/besu/cli/options/unstable/ChainPruningOptions.java

@@ -28,10 +28,17 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
   private static final String CHAIN_PRUNING_ENABLED_FLAG = "--Xchain-pruning-enabled";
   private static final String CHAIN_PRUNING_BLOCKS_RETAINED_FLAG =
       "--Xchain-pruning-blocks-retained";
+  private static final String CHAIN_PRUNING_BLOCKS_RETAINED_LIMIT_FLAG =
+      "--Xchain-pruning-blocks-retained-limit";
   private static final String CHAIN_PRUNING_FREQUENCY_FLAG = "--Xchain-pruning-frequency";
 
-  /** The constant DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED. */
+  /**
-  public static final long DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED = 7200;
+   * The "CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT" field sets the minimum limit for the
+   * "CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED" value. For most networks, the default value of this
+   * limit is the safest. Reducing this value requires careful consideration and understanding of
+   * the potential implications. Lowering this limit may have unintended side effects.
+   */
+  public static final long CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT = 7200;
 
   /** The constant DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY. */
   public static final int DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY = 256;
@@ -47,11 +54,21 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
       hidden = true,
       names = {CHAIN_PRUNING_BLOCKS_RETAINED_FLAG},
       description =
-          "The number of recent blocks for which to keep the chain data. Must be >= "
+          "The number of recent blocks for which to keep the chain data. Should be >= "
-              + DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED
+              + CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT
+              + " (default: ${DEFAULT-VALUE})")
+  private final Long chainDataPruningBlocksRetained = CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
+
+  @CommandLine.Option(
+      hidden = true,
+      names = {CHAIN_PRUNING_BLOCKS_RETAINED_LIMIT_FLAG},
+      description =
+          "Allows setting the limit below which no more blocks can be pruned. This prevents setting a value lower than this for "
+              + CHAIN_PRUNING_BLOCKS_RETAINED_FLAG
+              + ". This flag should be used with caution as reducing the limit may have unintended side effects."
               + " (default: ${DEFAULT-VALUE})")
-  private final Long chainDataPruningBlocksRetained =
+  private final Long chainDataPruningBlocksRetainedLimit =
-      DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED;
+      CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
 
   @CommandLine.Option(
       hidden = true,
@@ -91,11 +108,21 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
     return chainDataPruningBlocksRetained;
   }
 
+  /**
+   * Get the configured number of retained blocks for chain pruning.
+   *
+   * @return the number of retained blocks
+   */
+  public Long getChainDataPruningBlocksRetainedLimit() {
+    return chainDataPruningBlocksRetainedLimit;
+  }
+
   @Override
   public ChainPrunerConfiguration toDomainObject() {
     return new ChainPrunerConfiguration(
         chainDataPruningEnabled,
         chainDataPruningBlocksRetained,
+        chainDataPruningBlocksRetainedLimit,
         chainDataPruningBlocksFrequency.getValue());
   }
 
@@ -106,6 +133,8 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
         chainDataPruningEnabled.toString(),
         CHAIN_PRUNING_BLOCKS_RETAINED_FLAG,
         chainDataPruningBlocksRetained.toString(),
+        CHAIN_PRUNING_BLOCKS_RETAINED_LIMIT_FLAG,
+        chainDataPruningBlocksRetainedLimit.toString(),
         CHAIN_PRUNING_FREQUENCY_FLAG,
         chainDataPruningBlocksFrequency.toString());
   }

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

@@ -610,16 +610,6 @@ public abstract class BesuControllerBuilder implements MiningParameterOverrides
             blockchain, worldStateArchive, protocolSchedule, this::createConsensusContext);
     validateContext(protocolContext);
 
-    if (chainPrunerConfiguration.getChainPruningEnabled()) {
-      final ChainDataPruner chainDataPruner = createChainPruner(blockchainStorage);
-      blockchain.observeBlockAdded(chainDataPruner);
-      LOG.info(
-          "Chain data pruning enabled with recent blocks retained to be: "
-              + chainPrunerConfiguration.getChainPruningBlocksRetained()
-              + " and frequency to be: "
-              + chainPrunerConfiguration.getChainPruningBlocksFrequency());
-    }
-
     protocolSchedule.setPublicWorldStateArchiveForPrivacyBlockProcessor(
         protocolContext.getWorldStateArchive());
 
@@ -668,6 +658,16 @@ public abstract class BesuControllerBuilder implements MiningParameterOverrides
     final boolean fullSyncDisabled = !SyncMode.isFullSync(syncConfig.getSyncMode());
     final SyncState syncState = new SyncState(blockchain, ethPeers, fullSyncDisabled, checkpoint);
 
+    if (chainPrunerConfiguration.getChainPruningEnabled()) {
+      final ChainDataPruner chainDataPruner = createChainPruner(blockchainStorage);
+      blockchain.observeBlockAdded(chainDataPruner);
+      LOG.info(
+          "Chain data pruning enabled with recent blocks retained to be: "
+              + chainPrunerConfiguration.getChainPruningBlocksRetained()
+              + " and frequency to be: "
+              + chainPrunerConfiguration.getChainPruningBlocksFrequency());
+    }
+
     final TransactionPool transactionPool =
         TransactionPoolFactory.createTransactionPool(
             protocolSchedule,

besu/src/main/java/org/hyperledger/besu/services/BesuConfigurationImpl.java

@@ -14,6 +14,7 @@
  */
 package org.hyperledger.besu.services;
 
+import org.hyperledger.besu.cli.options.stable.JsonRpcHttpOptions;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
@@ -21,13 +22,18 @@ import org.hyperledger.besu.plugin.services.BesuConfiguration;
 import org.hyperledger.besu.plugin.services.storage.DataStorageFormat;
 
 import java.nio.file.Path;
+import java.util.Optional;
 
 /** A concrete implementation of BesuConfiguration which is used in Besu plugin framework. */
 public class BesuConfigurationImpl implements BesuConfiguration {
   private Path storagePath;
   private Path dataPath;
   private DataStorageConfiguration dataStorageConfiguration;
-  private MiningParameters miningParameters;
+
+  // defaults
+  private MiningParameters miningParameters = MiningParameters.newDefault();
+  private Optional<String> rpcHttpHost = Optional.of("http://localhost");
+  private Optional<Integer> rpcHttpPort = Optional.of(8545);
 
   /** Default Constructor. */
   public BesuConfigurationImpl() {}
@@ -38,17 +44,49 @@ public class BesuConfigurationImpl implements BesuConfiguration {
    * @param dataPath The Path representing data folder
    * @param storagePath The path representing storage folder
    * @param dataStorageConfiguration The data storage configuration
-   * @param miningParameters The mining parameters
+   * @return BesuConfigurationImpl instance
    */
-  public void init(
+  public BesuConfigurationImpl init(
       final Path dataPath,
       final Path storagePath,
-      final DataStorageConfiguration dataStorageConfiguration,
+      final DataStorageConfiguration dataStorageConfiguration) {
-      final MiningParameters miningParameters) {
     this.dataPath = dataPath;
     this.storagePath = storagePath;
     this.dataStorageConfiguration = dataStorageConfiguration;
+    return this;
+  }
+
+  /**
+   * Set the mining parameters
+   *
+   * @param miningParameters configured mining parameters
+   * @return BesuConfigurationImpl instance
+   */
+  public BesuConfigurationImpl withMiningParameters(final MiningParameters miningParameters) {
     this.miningParameters = miningParameters;
+    return this;
+  }
+
+  /**
+   * Set the RPC http options
+   *
+   * @param rpcHttpOptions configured rpc http options
+   * @return BesuConfigurationImpl instance
+   */
+  public BesuConfigurationImpl withJsonRpcHttpOptions(final JsonRpcHttpOptions rpcHttpOptions) {
+    this.rpcHttpHost = Optional.ofNullable(rpcHttpOptions.getRpcHttpHost());
+    this.rpcHttpPort = Optional.ofNullable(rpcHttpOptions.getRpcHttpPort());
+    return this;
+  }
+
+  @Override
+  public Optional<String> getRpcHttpHost() {
+    return rpcHttpHost;
+  }
+
+  @Override
+  public Optional<Integer> getRpcHttpPort() {
+    return rpcHttpPort;
   }
 
   @Override

besu/src/main/java/org/hyperledger/besu/services/BesuEventsImpl.java

@@ -114,6 +114,11 @@ public class BesuEventsImpl implements BesuEvents {
     blockchain.removeObserver(listenerIdentifier);
   }
 
+  @Override
+  public long addInitialSyncCompletionListener(final InitialSyncCompletionListener listener) {
+    return syncState.subscribeCompletionReached(listener);
+  }
+
   @Override
   public long addTransactionAddedListener(final TransactionAddedListener listener) {
     return transactionPool.subscribePendingTransactions(listener::onTransactionAdded);

besu/src/main/java/org/hyperledger/besu/services/BesuPluginContextImpl.java

@@ -262,6 +262,25 @@ public class BesuPluginContextImpl implements BesuContext, PluginVersionsProvide
     state = Lifecycle.BEFORE_MAIN_LOOP_FINISHED;
   }
 
+  /** Execute all plugin setup code after external services. */
+  public void afterExternalServicesMainLoop() {
+    checkState(
+        state == Lifecycle.BEFORE_MAIN_LOOP_FINISHED,
+        "BesuContext should be in state %s but it was in %s",
+        Lifecycle.BEFORE_MAIN_LOOP_FINISHED,
+        state);
+    final Iterator<BesuPlugin> pluginsIterator = registeredPlugins.iterator();
+
+    while (pluginsIterator.hasNext()) {
+      final BesuPlugin plugin = pluginsIterator.next();
+      try {
+        plugin.afterExternalServicePostMainLoop();
+      } finally {
+        pluginsIterator.remove();
+      }
+    }
+  }
+
   /** Stop plugins. */
   public void stopPlugins() {
     checkState(

besu/src/main/java/org/hyperledger/besu/services/BlockchainServiceImpl.java

@@ -17,17 +17,22 @@ package org.hyperledger.besu.services;
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.ProtocolContext;
-import org.hyperledger.besu.ethereum.core.BlockBody;
+import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
+import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.BaseFeeMarket;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
 import org.hyperledger.besu.plugin.Unstable;
+import org.hyperledger.besu.plugin.data.BlockBody;
 import org.hyperledger.besu.plugin.data.BlockContext;
 import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.TransactionReceipt;
 import org.hyperledger.besu.plugin.services.BlockchainService;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 /** The Blockchain service implementation. */
 @Unstable
@@ -35,9 +40,7 @@ public class BlockchainServiceImpl implements BlockchainService {
 
   private ProtocolContext protocolContext;
   private ProtocolSchedule protocolSchedule;
-
+  private MutableBlockchain blockchain;
-  /** Create a new instance */
-  public BlockchainServiceImpl() {}
 
   /**
    * Instantiates a new Blockchain service.
@@ -48,6 +51,7 @@ public class BlockchainServiceImpl implements BlockchainService {
   public void init(final ProtocolContext protocolContext, final ProtocolSchedule protocolSchedule) {
     this.protocolContext = protocolContext;
     this.protocolSchedule = protocolSchedule;
+    this.blockchain = protocolContext.getBlockchain();
   }
 
   /**
@@ -91,6 +95,43 @@ public class BlockchainServiceImpl implements BlockchainService {
                     feeMarket.targetGasUsed(chainHeadHeader)));
   }
 
+  @Override
+  public Optional<List<TransactionReceipt>> getReceiptsByBlockHash(final Hash blockHash) {
+    return blockchain
+        .getTxReceipts(blockHash)
+        .map(
+            list -> list.stream().map(TransactionReceipt.class::cast).collect(Collectors.toList()));
+  }
+
+  @Override
+  public void storeBlock(
+      final BlockHeader blockHeader,
+      final BlockBody blockBody,
+      final List<TransactionReceipt> receipts) {
+    final org.hyperledger.besu.ethereum.core.BlockHeader coreHeader =
+        (org.hyperledger.besu.ethereum.core.BlockHeader) blockHeader;
+    final org.hyperledger.besu.ethereum.core.BlockBody coreBody =
+        (org.hyperledger.besu.ethereum.core.BlockBody) blockBody;
+    final List<org.hyperledger.besu.ethereum.core.TransactionReceipt> coreReceipts =
+        receipts.stream()
+            .map(org.hyperledger.besu.ethereum.core.TransactionReceipt.class::cast)
+            .toList();
+    blockchain.unsafeImportBlock(
+        new Block(coreHeader, coreBody),
+        coreReceipts,
+        Optional.ofNullable(blockchain.calculateTotalDifficulty(coreHeader)));
+  }
+
+  @Override
+  public Optional<Hash> getSafeBlock() {
+    return blockchain.getSafeBlock();
+  }
+
+  @Override
+  public Optional<Hash> getFinalizedBlock() {
+    return blockchain.getFinalized();
+  }
+
   private static BlockContext blockContext(
       final Supplier<BlockHeader> blockHeaderSupplier,
       final Supplier<BlockBody> blockBodySupplier) {

besu/src/main/java/org/hyperledger/besu/services/P2PServiceImpl.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services;
+
+import org.hyperledger.besu.ethereum.p2p.network.P2PNetwork;
+import org.hyperledger.besu.plugin.services.p2p.P2PService;
+
+/** Service to enable and disable P2P discovery. */
+public class P2PServiceImpl implements P2PService {
+
+  private final P2PNetwork p2PNetwork;
+
+  /**
+   * Creates a new P2PServiceImpl.
+   *
+   * @param p2PNetwork the P2P network to enable and disable.
+   */
+  public P2PServiceImpl(final P2PNetwork p2PNetwork) {
+    this.p2PNetwork = p2PNetwork;
+  }
+
+  /** Enables P2P discovery. */
+  @Override
+  public void enableDiscovery() {
+    p2PNetwork.start();
+  }
+
+  @Override
+  public void disableDiscovery() {
+    p2PNetwork.stop();
+  }
+}

besu/src/main/java/org/hyperledger/besu/services/RlpConverterServiceImpl.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services;
+
+import org.hyperledger.besu.ethereum.core.BlockHeaderFunctions;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ScheduleBasedBlockHeaderFunctions;
+import org.hyperledger.besu.ethereum.rlp.RLP;
+import org.hyperledger.besu.plugin.data.BlockBody;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.TransactionReceipt;
+import org.hyperledger.besu.plugin.services.rlp.RlpConverterService;
+
+import org.apache.tuweni.bytes.Bytes;
+
+/** RLP Serialiaztion/Deserialization service. */
+public class RlpConverterServiceImpl implements RlpConverterService {
+
+  private final BlockHeaderFunctions blockHeaderFunctions;
+
+  /**
+   * Constructor for RlpConverterServiceImpl.
+   *
+   * @param protocolSchedule the protocol schedule.
+   */
+  public RlpConverterServiceImpl(final ProtocolSchedule protocolSchedule) {
+    this.blockHeaderFunctions = ScheduleBasedBlockHeaderFunctions.create(protocolSchedule);
+  }
+
+  @Override
+  public BlockHeader buildHeaderFromRlp(final Bytes rlp) {
+    return org.hyperledger.besu.ethereum.core.BlockHeader.readFrom(
+        RLP.input(rlp), blockHeaderFunctions);
+  }
+
+  @Override
+  public BlockBody buildBodyFromRlp(final Bytes rlp) {
+    return org.hyperledger.besu.ethereum.core.BlockBody.readWrappedBodyFrom(
+        RLP.input(rlp), blockHeaderFunctions);
+  }
+
+  @Override
+  public TransactionReceipt buildReceiptFromRlp(final Bytes rlp) {
+    return org.hyperledger.besu.ethereum.core.TransactionReceipt.readFrom(RLP.input(rlp));
+  }
+
+  @Override
+  public Bytes buildRlpFromHeader(final BlockHeader blockHeader) {
+    return RLP.encode(
+        org.hyperledger.besu.ethereum.core.BlockHeader.convertPluginBlockHeader(
+                blockHeader, blockHeaderFunctions)
+            ::writeTo);
+  }
+
+  @Override
+  public Bytes buildRlpFromBody(final BlockBody blockBody) {
+    return RLP.encode(
+        rlpOutput ->
+            ((org.hyperledger.besu.ethereum.core.BlockBody) blockBody)
+                .writeWrappedBodyTo(rlpOutput));
+  }
+
+  @Override
+  public Bytes buildRlpFromReceipt(final TransactionReceipt receipt) {
+    return RLP.encode(
+        rlpOutput ->
+            ((org.hyperledger.besu.ethereum.core.TransactionReceipt) receipt)
+                .writeToForNetwork(rlpOutput));
+  }
+}

besu/src/main/java/org/hyperledger/besu/services/SynchronizationServiceImpl.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services;
+
+import org.hyperledger.besu.consensus.merge.MergeContext;
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.ProtocolContext;
+import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockImporter;
+import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.ethereum.mainnet.HeaderValidationMode;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.DiffBasedWorldStateProvider;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedWorldStateKeyValueStorage;
+import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
+import org.hyperledger.besu.plugin.data.BlockBody;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.services.sync.SynchronizationService;
+
+import java.util.Optional;
+
+import org.apache.tuweni.bytes.Bytes;
+import org.apache.tuweni.bytes.Bytes32;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Synchronization service. */
+public class SynchronizationServiceImpl implements SynchronizationService {
+
+  private static final Logger LOG = LoggerFactory.getLogger(SynchronizationServiceImpl.class);
+
+  private final ProtocolContext protocolContext;
+  private final ProtocolSchedule protocolSchedule;
+
+  private final SyncState syncState;
+  private final Optional<DiffBasedWorldStateProvider> worldStateArchive;
+
+  /**
+   * Constructor for SynchronizationServiceImpl.
+   *
+   * @param protocolContext protocol context
+   * @param protocolSchedule protocol schedule
+   * @param syncState sync state
+   * @param worldStateArchive world state archive
+   */
+  public SynchronizationServiceImpl(
+      final ProtocolContext protocolContext,
+      final ProtocolSchedule protocolSchedule,
+      final SyncState syncState,
+      final WorldStateArchive worldStateArchive) {
+    this.protocolContext = protocolContext;
+    this.protocolSchedule = protocolSchedule;
+    this.syncState = syncState;
+    this.worldStateArchive =
+        Optional.ofNullable(worldStateArchive)
+            .filter(z -> z instanceof DiffBasedWorldStateProvider)
+            .map(DiffBasedWorldStateProvider.class::cast);
+  }
+
+  @Override
+  public void fireNewUnverifiedForkchoiceEvent(
+      final Hash head, final Hash safeBlock, final Hash finalizedBlock) {
+    final MergeContext mergeContext = protocolContext.getConsensusContext(MergeContext.class);
+    if (mergeContext != null) {
+      mergeContext.fireNewUnverifiedForkchoiceEvent(head, safeBlock, finalizedBlock);
+      protocolContext.getBlockchain().setFinalized(finalizedBlock);
+      protocolContext.getBlockchain().setSafeBlock(safeBlock);
+    } else {
+      LOG.atWarn()
+          .setMessage(
+              "The merge context is unavailable, hence the fork choice event cannot be triggered")
+          .log();
+    }
+  }
+
+  @Override
+  public boolean setHead(final BlockHeader blockHeader, final BlockBody blockBody) {
+    final BlockImporter blockImporter =
+        protocolSchedule
+            .getByBlockHeader((org.hyperledger.besu.ethereum.core.BlockHeader) blockHeader)
+            .getBlockImporter();
+    return blockImporter
+        .importBlock(
+            protocolContext,
+            new Block(
+                (org.hyperledger.besu.ethereum.core.BlockHeader) blockHeader,
+                (org.hyperledger.besu.ethereum.core.BlockBody) blockBody),
+            HeaderValidationMode.SKIP_DETACHED)
+        .isImported();
+  }
+
+  @Override
+  public boolean setHeadUnsafe(final BlockHeader blockHeader, final BlockBody blockBody) {
+    final org.hyperledger.besu.ethereum.core.BlockHeader coreHeader =
+        (org.hyperledger.besu.ethereum.core.BlockHeader) blockHeader;
+
+    final MutableBlockchain blockchain = protocolContext.getBlockchain();
+
+    if (worldStateArchive.flatMap(archive -> archive.getMutable(coreHeader, true)).isPresent()) {
+      if (coreHeader.getParentHash().equals(blockchain.getChainHeadHash())) {
+        LOG.atDebug()
+            .setMessage(
+                "Forwarding chain head to the block {} saved from a previous newPayload invocation")
+            .addArgument(coreHeader::toLogString)
+            .log();
+        return blockchain.forwardToBlock(coreHeader);
+      } else {
+        LOG.atDebug()
+            .setMessage("New head {} is a chain reorg, rewind chain head to it")
+            .addArgument(coreHeader::toLogString)
+            .log();
+        return blockchain.rewindToBlock(coreHeader.getBlockHash());
+      }
+    } else {
+      LOG.atWarn()
+          .setMessage("The world state is unavailable, setting of head cannot be performed.")
+          .log();
+    }
+    return false;
+  }
+
+  @Override
+  public boolean isInitialSyncPhaseDone() {
+    return syncState.isInitialSyncPhaseDone();
+  }
+
+  @Override
+  public void disableWorldStateTrie() {
+    // TODO maybe find a best way in the future to delete and disable trie
+    worldStateArchive.ifPresent(
+        archive -> {
+          archive.getDefaultWorldStateConfig().setTrieDisabled(true);
+          final DiffBasedWorldStateKeyValueStorage worldStateStorage =
+              archive.getWorldStateKeyValueStorage();
+          final Optional<Hash> worldStateBlockHash = worldStateStorage.getWorldStateBlockHash();
+          final Optional<Bytes> worldStateRootHash = worldStateStorage.getWorldStateRootHash();
+          if (worldStateRootHash.isPresent() && worldStateBlockHash.isPresent()) {
+            worldStateStorage.clearTrie();
+            // keep root and block hash in the trie branch
+            final DiffBasedWorldStateKeyValueStorage.Updater updater = worldStateStorage.updater();
+            updater.saveWorldState(
+                worldStateBlockHash.get(), Bytes32.wrap(worldStateRootHash.get()), Bytes.EMPTY);
+            updater.commit();
+
+            // currently only bonsai needs an explicit upgrade to full flat db
+            if (worldStateStorage instanceof BonsaiWorldStateKeyValueStorage bonsaiStorage) {
+              bonsaiStorage.upgradeToFullFlatDbMode();
+            }
+          }
+        });
+  }
+}

besu/src/main/java/org/hyperledger/besu/services/TransactionPoolServiceImpl.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services;
+
+import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
+import org.hyperledger.besu.plugin.services.transactionpool.TransactionPoolService;
+
+/** Service to enable and disable the transaction pool. */
+public class TransactionPoolServiceImpl implements TransactionPoolService {
+
+  private final TransactionPool transactionPool;
+
+  /**
+   * Creates a new TransactionPoolServiceImpl.
+   *
+   * @param transactionPool the transaction pool to control
+   */
+  public TransactionPoolServiceImpl(final TransactionPool transactionPool) {
+    this.transactionPool = transactionPool;
+  }
+
+  @Override
+  public void disableTransactionPool() {
+    transactionPool.setDisabled();
+  }
+
+  @Override
+  public void enableTransactionPool() {
+    transactionPool.setEnabled();
+  }
+}

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

@@ -137,7 +137,9 @@ public class PrivacyTest {
       final DataStorageConfiguration dataStorageConfiguration,
       final MiningParameters miningParameters) {
     final var besuConfiguration = new BesuConfigurationImpl();
-    besuConfiguration.init(dataDir, dbDir, dataStorageConfiguration, miningParameters);
+    besuConfiguration
+        .init(dataDir, dbDir, dataStorageConfiguration)
+        .withMiningParameters(miningParameters);
     return new PrivacyKeyValueStorageProviderBuilder()
         .withStorageFactory(
             new RocksDBKeyValuePrivacyStorageFactory(

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

@@ -391,7 +391,9 @@ public final class RunnerTest {
       final DataStorageConfiguration dataStorageConfiguration,
       final MiningParameters miningParameters) {
     final var besuConfiguration = new BesuConfigurationImpl();
-    besuConfiguration.init(dataDir, dbDir, dataStorageConfiguration, miningParameters);
+    besuConfiguration
+        .init(dataDir, dbDir, dataStorageConfiguration)
+        .withMiningParameters(miningParameters);
     return new KeyValueStorageProviderBuilder()
         .withStorageFactory(
             new RocksDBKeyValueStorageFactory(

besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java

@@ -127,6 +127,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
 import org.mockito.Mock;
+import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -214,7 +215,7 @@ public abstract class CommandTestAbstract {
   @Mock protected TransactionSelectionServiceImpl txSelectionService;
   @Mock protected SecurityModuleServiceImpl securityModuleService;
   @Mock protected SecurityModule securityModule;
-  @Mock protected BesuConfigurationImpl commonPluginConfiguration;
+  @Spy protected BesuConfigurationImpl commonPluginConfiguration = new BesuConfigurationImpl();
   @Mock protected KeyValueStorageFactory rocksDBStorageFactory;
   @Mock protected PrivacyKeyValueStorageFactory rocksDBSPrivacyStorageFactory;
   @Mock protected PicoCLIOptions cliOptions;

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

@@ -30,7 +30,6 @@ import org.hyperledger.besu.consensus.common.bft.blockcreation.BftMiningCoordina
 import org.hyperledger.besu.ethereum.ConsensusContext;
 import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
-import org.hyperledger.besu.ethereum.blockcreation.NoopMiningCoordinator;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
@@ -68,7 +67,7 @@ public class ConsensusScheduleBesuControllerBuilderTest {
   private @Mock ProtocolSchedule protocolSchedule1;
   private @Mock ProtocolSchedule protocolSchedule2;
   private @Mock ProtocolSchedule protocolSchedule3;
-  private @Mock NoopMiningCoordinator miningCoordinator1;
+  private @Mock MiningCoordinator miningCoordinator1;
   private @Mock BftMiningCoordinator miningCoordinator2;
 
   @Test

besu/src/test/java/org/hyperledger/besu/services/RlpConverterServiceImplTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.hyperledger.besu.datatypes.BlobGas;
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.datatypes.Wei;
+import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
+import org.hyperledger.besu.ethereum.core.ProtocolScheduleFixture;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+
+import org.apache.tuweni.bytes.Bytes;
+import org.junit.jupiter.api.Test;
+
+public class RlpConverterServiceImplTest {
+
+  @Test
+  public void testBuildRlpFromHeader() {
+    // Arrange
+    RlpConverterServiceImpl rlpConverterServiceImpl =
+        new RlpConverterServiceImpl(ProtocolScheduleFixture.MAINNET);
+    // header with cancun fields
+    BlockHeader header =
+        new BlockHeaderTestFixture()
+            .timestamp(1710338135 + 1)
+            .baseFeePerGas(Wei.of(1000))
+            .requestsRoot(Hash.ZERO)
+            .withdrawalsRoot(Hash.ZERO)
+            .blobGasUsed(500L)
+            .excessBlobGas(BlobGas.of(500L))
+            .buildHeader();
+
+    Bytes rlpBytes = rlpConverterServiceImpl.buildRlpFromHeader(header);
+    BlockHeader deserialized = rlpConverterServiceImpl.buildHeaderFromRlp(rlpBytes);
+    // Assert
+    assertThat(header).isEqualTo(deserialized);
+    assertThat(header.getBlobGasUsed()).isEqualTo(deserialized.getBlobGasUsed());
+    assertThat(header.getExcessBlobGas()).isEqualTo(deserialized.getExcessBlobGas());
+  }
+}

consensus/common/src/test/java/org/hyperledger/besu/consensus/common/MigratingMiningCoordinatorTest.java

@@ -30,7 +30,6 @@ import org.hyperledger.besu.consensus.common.bft.blockcreation.BftBlockCreatorFa
 import org.hyperledger.besu.consensus.common.bft.blockcreation.BftMiningCoordinator;
 import org.hyperledger.besu.consensus.common.bft.statemachine.BftEventHandler;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
-import org.hyperledger.besu.ethereum.blockcreation.NoopMiningCoordinator;
 import org.hyperledger.besu.ethereum.chain.BlockAddedEvent;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.core.Block;
@@ -122,7 +121,7 @@ public class MigratingMiningCoordinatorTest {
 
   @Test
   public void onBlockAddedShouldNotDelegateWhenDelegateIsNoop() {
-    NoopMiningCoordinator mockNoopCoordinator = mock(NoopMiningCoordinator.class);
+    MiningCoordinator mockNoopCoordinator = mock(MiningCoordinator.class);
     coordinatorSchedule = createCoordinatorSchedule(mockNoopCoordinator, coordinator2);
     when(blockHeader.getNumber()).thenReturn(GENESIS_BLOCK_NUMBER);
 

datatypes/src/main/java/org/hyperledger/besu/datatypes/BlobGas.java

@@ -16,6 +16,7 @@ package org.hyperledger.besu.datatypes;
 
 import java.math.BigInteger;
 
+import com.fasterxml.jackson.annotation.JsonValue;
 import org.apache.tuweni.bytes.Bytes;
 import org.apache.tuweni.units.bigints.BaseUInt64Value;
 import org.apache.tuweni.units.bigints.UInt64;
@@ -123,6 +124,7 @@ public final class BlobGas extends BaseUInt64Value<BlobGas> implements Quantity
     return toBigInteger();
   }
 
+  @JsonValue
   @Override
   public String toHexString() {
     return super.toHexString();
@@ -140,6 +142,6 @@ public final class BlobGas extends BaseUInt64Value<BlobGas> implements Quantity
    * @return the blob gas
    */
   public static BlobGas fromQuantity(final Quantity quantity) {
-    return BlobGas.wrap((Bytes) quantity);
+    return BlobGas.of(quantity.getAsBigInteger());
   }
 }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/ChainPrunerConfiguration.java

@@ -16,15 +16,20 @@ package org.hyperledger.besu.ethereum.chain;
 
 public class ChainPrunerConfiguration {
   public static final ChainPrunerConfiguration DEFAULT =
-      new ChainPrunerConfiguration(false, 7200, 256);
+      new ChainPrunerConfiguration(false, 7200, 7200, 256);
   private final boolean enabled;
   private final long blocksRetained;
   private final long blocksFrequency;
+  private final long blocksRetainedLimit;
 
   public ChainPrunerConfiguration(
-      final boolean enabled, final long blocksRetained, final long blocksFrequency) {
+      final boolean enabled,
+      final long blocksRetained,
+      final long blocksRetainedLimit,
+      final long blocksFrequency) {
     this.enabled = enabled;
     this.blocksRetained = blocksRetained;
+    this.blocksRetainedLimit = blocksRetainedLimit;
     this.blocksFrequency = blocksFrequency;
   }
 
@@ -32,6 +37,10 @@ public class ChainPrunerConfiguration {
     return blocksRetained;
   }
 
+  public long getBlocksRetainedLimit() {
+    return blocksRetainedLimit;
+  }
+
   public boolean getChainPruningEnabled() {
     return enabled;
   }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/DefaultBlockchain.java

@@ -469,7 +469,8 @@ public class DefaultBlockchain implements MutableBlockchain {
     updater.commit();
   }
 
-  private Difficulty calculateTotalDifficulty(final BlockHeader blockHeader) {
+  @Override
+  public Difficulty calculateTotalDifficulty(final BlockHeader blockHeader) {
     if (blockHeader.getNumber() == BlockHeader.GENESIS_BLOCK_NUMBER) {
       return blockHeader.getDifficulty();
     }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/MutableBlockchain.java

@@ -56,6 +56,8 @@ public interface MutableBlockchain extends Blockchain {
 
   void unsafeSetChainHead(final BlockHeader blockHeader, final Difficulty totalDifficulty);
 
+  Difficulty calculateTotalDifficulty(final BlockHeader blockHeader);
+
   /**
    * Rolls back the canonical chainhead to the specified block number.
    *

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

@@ -309,7 +309,7 @@ public class BlockHeader extends SealableBlockHeader
             .map(h -> Hash.fromHexString(h.toHexString()))
             .orElse(null),
         pluginBlockHeader.getBlobGasUsed().map(Long::longValue).orElse(null),
-        pluginBlockHeader.getExcessBlobGas().map(BlobGas::fromQuantity).orElse(null),
+        pluginBlockHeader.getExcessBlobGas().map(BlobGas.class::cast).orElse(null),
         pluginBlockHeader.getParentBeaconBlockRoot().orElse(null),
         pluginBlockHeader
             .getRequestsRoot()

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/common/GenesisWorldStateProvider.java

@@ -22,6 +22,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.cache.NoOpBonsaiCache
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldState;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.NoOpTrieLogManager;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.trie.forest.storage.ForestWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.forest.worldview.ForestMutableWorldState;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
@@ -72,7 +73,8 @@ public class GenesisWorldStateProvider {
         bonsaiCachedMerkleTrieLoader,
         new NoOpBonsaiCachedWorldStorageManager(bonsaiWorldStateKeyValueStorage),
         new NoOpTrieLogManager(),
-        EvmConfiguration.DEFAULT);
+        EvmConfiguration.DEFAULT,
+        new DiffBasedWorldStateConfig());
   }
 
   /**

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/BonsaiWorldStateProvider.java

@@ -24,6 +24,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldSt
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldState;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.DiffBasedWorldStateProvider;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogManager;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.trie.patricia.StoredMerklePatriciaTrie;
 import org.hyperledger.besu.ethereum.worldstate.StateTrieAccountValue;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
@@ -54,8 +55,11 @@ public class BonsaiWorldStateProvider extends DiffBasedWorldStateProvider {
     super(worldStateKeyValueStorage, blockchain, maxLayersToLoad, pluginContext);
     this.bonsaiCachedMerkleTrieLoader = bonsaiCachedMerkleTrieLoader;
     provideCachedWorldStorageManager(
-        new BonsaiCachedWorldStorageManager(this, worldStateKeyValueStorage));
+        new BonsaiCachedWorldStorageManager(
-    loadPersistedState(new BonsaiWorldState(this, worldStateKeyValueStorage, evmConfiguration));
+            this, worldStateKeyValueStorage, this::cloneBonsaiWorldStateConfig));
+    loadPersistedState(
+        new BonsaiWorldState(
+            this, worldStateKeyValueStorage, evmConfiguration, defaultWorldStateConfig));
   }
 
   @VisibleForTesting
@@ -69,14 +73,16 @@ public class BonsaiWorldStateProvider extends DiffBasedWorldStateProvider {
     super(worldStateKeyValueStorage, blockchain, trieLogManager);
     this.bonsaiCachedMerkleTrieLoader = bonsaiCachedMerkleTrieLoader;
     provideCachedWorldStorageManager(bonsaiCachedWorldStorageManager);
-    loadPersistedState(new BonsaiWorldState(this, worldStateKeyValueStorage, evmConfiguration));
+    loadPersistedState(
+        new BonsaiWorldState(
+            this, worldStateKeyValueStorage, evmConfiguration, defaultWorldStateConfig));
   }
 
   public BonsaiCachedMerkleTrieLoader getCachedMerkleTrieLoader() {
     return bonsaiCachedMerkleTrieLoader;
   }
 
-  private BonsaiWorldStateKeyValueStorage getWorldStateKeyValueStorage() {
+  private BonsaiWorldStateKeyValueStorage getBonsaiWorldStateKeyValueStorage() {
     return (BonsaiWorldStateKeyValueStorage) worldStateKeyValueStorage;
   }
 
@@ -89,13 +95,13 @@ public class BonsaiWorldStateProvider extends DiffBasedWorldStateProvider {
   public void prepareStateHealing(final Address address, final Bytes location) {
     final Set<Bytes> keysToDelete = new HashSet<>();
     final BonsaiWorldStateKeyValueStorage.Updater updater =
-        getWorldStateKeyValueStorage().updater();
+        getBonsaiWorldStateKeyValueStorage().updater();
     final Hash accountHash = address.addressHash();
     final StoredMerklePatriciaTrie<Bytes, Bytes> accountTrie =
         new StoredMerklePatriciaTrie<>(
             (l, h) -> {
               final Optional<Bytes> node =
-                  getWorldStateKeyValueStorage().getAccountStateTrieNode(l, h);
+                  getBonsaiWorldStateKeyValueStorage().getAccountStateTrieNode(l, h);
               if (node.isPresent()) {
                 keysToDelete.add(l);
               }
@@ -115,7 +121,7 @@ public class BonsaiWorldStateProvider extends DiffBasedWorldStateProvider {
                     new StoredMerklePatriciaTrie<>(
                         (l, h) -> {
                           Optional<Bytes> node =
-                              getWorldStateKeyValueStorage()
+                              getBonsaiWorldStateKeyValueStorage()
                                   .getAccountStorageTrieNode(accountHash, l, h);
                           if (node.isPresent()) {
                             keysToDelete.add(Bytes.concatenate(accountHash, l));
@@ -139,6 +145,10 @@ public class BonsaiWorldStateProvider extends DiffBasedWorldStateProvider {
     keysToDelete.forEach(updater::removeAccountStateTrieNode);
     updater.commit();
 
-    getWorldStateKeyValueStorage().downgradeToPartialFlatDbMode();
+    getBonsaiWorldStateKeyValueStorage().downgradeToPartialFlatDbMode();
+  }
+
+  private DiffBasedWorldStateConfig cloneBonsaiWorldStateConfig() {
+    return new DiffBasedWorldStateConfig(defaultWorldStateConfig);
   }
 }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/cache/BonsaiCachedWorldStorageManager.java

@@ -23,14 +23,18 @@ import org.hyperledger.besu.ethereum.trie.diffbased.common.DiffBasedWorldStatePr
 import org.hyperledger.besu.ethereum.trie.diffbased.common.cache.DiffBasedCachedWorldStorageManager;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 
+import java.util.function.Supplier;
+
 public class BonsaiCachedWorldStorageManager extends DiffBasedCachedWorldStorageManager {
 
   public BonsaiCachedWorldStorageManager(
       final BonsaiWorldStateProvider archive,
-      final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage) {
+      final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage,
-    super(archive, worldStateKeyValueStorage);
+      final Supplier<DiffBasedWorldStateConfig> defaultBonsaiWorldStateConfigSupplier) {
+    super(archive, worldStateKeyValueStorage, defaultBonsaiWorldStateConfigSupplier);
   }
 
   @Override
@@ -41,7 +45,8 @@ public class BonsaiCachedWorldStorageManager extends DiffBasedCachedWorldStorage
     return new BonsaiWorldState(
         (BonsaiWorldStateProvider) archive,
         (BonsaiWorldStateKeyValueStorage) worldStateKeyValueStorage,
-        evmConfiguration);
+        evmConfiguration,
+        defaultBonsaiWorldStateConfigSupplier.get());
   }
 
   @Override

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/cache/NoOpBonsaiCachedWorldStorageManager.java

@@ -18,6 +18,7 @@ import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 
 import java.util.Optional;
 import java.util.function.Function;
@@ -26,7 +27,7 @@ public class NoOpBonsaiCachedWorldStorageManager extends BonsaiCachedWorldStorag
 
   public NoOpBonsaiCachedWorldStorageManager(
       final BonsaiWorldStateKeyValueStorage bonsaiWorldStateKeyValueStorage) {
-    super(null, bonsaiWorldStateKeyValueStorage);
+    super(null, bonsaiWorldStateKeyValueStorage, DiffBasedWorldStateConfig::new);
   }
 
   @Override

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/storage/BonsaiSnapshotWorldStateKeyValueStorage.java

@@ -203,6 +203,15 @@ public class BonsaiSnapshotWorldStateKeyValueStorage extends BonsaiWorldStateKey
     }
   }
 
+  @Override
+  public void onClearTrie() {
+    try {
+      doClose();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   @Override
   protected synchronized void doClose() throws Exception {
     if (!isClosedGet()) {

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/worldview/BonsaiWorldState.java

@@ -23,6 +23,7 @@ import org.hyperledger.besu.datatypes.StorageSlotKey;
 import org.hyperledger.besu.ethereum.core.MutableWorldState;
 import org.hyperledger.besu.ethereum.trie.MerkleTrie;
 import org.hyperledger.besu.ethereum.trie.MerkleTrieException;
+import org.hyperledger.besu.ethereum.trie.NoOpMerkleTrie;
 import org.hyperledger.besu.ethereum.trie.NodeLoader;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.BonsaiAccount;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.BonsaiWorldStateProvider;
@@ -34,6 +35,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.common.cache.DiffBasedCached
 import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogManager;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.accumulator.DiffBasedWorldStateUpdateAccumulator;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.accumulator.preload.StorageConsumingMap;
 import org.hyperledger.besu.ethereum.trie.patricia.StoredMerklePatriciaTrie;
@@ -63,13 +65,15 @@ public class BonsaiWorldState extends DiffBasedWorldState {
   public BonsaiWorldState(
       final BonsaiWorldStateProvider archive,
       final BonsaiWorldStateKeyValueStorage worldStateKeyValueStorage,
-      final EvmConfiguration evmConfiguration) {
+      final EvmConfiguration evmConfiguration,
+      final DiffBasedWorldStateConfig diffBasedWorldStateConfig) {
     this(
         worldStateKeyValueStorage,
         archive.getCachedMerkleTrieLoader(),
         archive.getCachedWorldStorageManager(),
         archive.getTrieLogManager(),
-        evmConfiguration);
+        evmConfiguration,
+        diffBasedWorldStateConfig);
   }
 
   public BonsaiWorldState(
@@ -77,21 +81,32 @@ public class BonsaiWorldState extends DiffBasedWorldState {
       final BonsaiCachedMerkleTrieLoader bonsaiCachedMerkleTrieLoader,
       final DiffBasedCachedWorldStorageManager cachedWorldStorageManager,
       final TrieLogManager trieLogManager,
-      final EvmConfiguration evmConfiguration) {
+      final EvmConfiguration evmConfiguration,
-    super(worldStateKeyValueStorage, cachedWorldStorageManager, trieLogManager);
+      final DiffBasedWorldStateConfig diffBasedWorldStateConfig) {
+    super(
+        worldStateKeyValueStorage,
+        cachedWorldStorageManager,
+        trieLogManager,
+        diffBasedWorldStateConfig);
     this.bonsaiCachedMerkleTrieLoader = bonsaiCachedMerkleTrieLoader;
+    this.worldStateKeyValueStorage = worldStateKeyValueStorage;
     this.setAccumulator(
         new BonsaiWorldStateUpdateAccumulator(
             this,
             (addr, value) ->
                 bonsaiCachedMerkleTrieLoader.preLoadAccount(
-                    worldStateKeyValueStorage, worldStateRootHash, addr),
+                    getWorldStateStorage(), worldStateRootHash, addr),
             (addr, value) ->
-                bonsaiCachedMerkleTrieLoader.preLoadStorageSlot(
+                this.bonsaiCachedMerkleTrieLoader.preLoadStorageSlot(
                     getWorldStateStorage(), addr, value),
             evmConfiguration));
   }
 
+  @Override
+  public Optional<Bytes> getCode(@Nonnull final Address address, final Hash codeHash) {
+    return getWorldStateStorage().getCode(codeHash, address.addressHash());
+  }
+
   @Override
   public BonsaiWorldStateKeyValueStorage getWorldStateStorage() {
     return (BonsaiWorldStateKeyValueStorage) worldStateKeyValueStorage;
@@ -129,7 +144,7 @@ public class BonsaiWorldState extends DiffBasedWorldState {
     updateCode(maybeStateUpdater, worldStateUpdater);
 
     // next walk the account trie
-    final StoredMerklePatriciaTrie<Bytes, Bytes> accountTrie =
+    final MerkleTrie<Bytes, Bytes> accountTrie =
         createTrie(
             (location, hash) ->
                 bonsaiCachedMerkleTrieLoader.getAccountStateTrieNode(
@@ -157,7 +172,7 @@ public class BonsaiWorldState extends DiffBasedWorldState {
   private void updateTheAccounts(
       final Optional<BonsaiWorldStateKeyValueStorage.Updater> maybeStateUpdater,
       final BonsaiWorldStateUpdateAccumulator worldStateUpdater,
-      final StoredMerklePatriciaTrie<Bytes, Bytes> accountTrie) {
+      final MerkleTrie<Bytes, Bytes> accountTrie) {
     for (final Map.Entry<Address, DiffBasedValue<BonsaiAccount>> accountUpdate :
         worldStateUpdater.getAccountsToUpdate().entrySet()) {
       final Bytes accountKey = accountUpdate.getKey();
@@ -234,7 +249,7 @@ public class BonsaiWorldState extends DiffBasedWorldState {
                   || worldStateUpdater.getStorageToClear().contains(updatedAddress))
               ? Hash.EMPTY_TRIE_HASH
               : accountOriginal.getStorageRoot();
-      final StoredMerklePatriciaTrie<Bytes, Bytes> storageTrie =
+      final MerkleTrie<Bytes, Bytes> storageTrie =
           createTrie(
               (location, key) ->
                   bonsaiCachedMerkleTrieLoader.getAccountStorageTrieNode(
@@ -277,8 +292,11 @@ public class BonsaiWorldState extends DiffBasedWorldState {
                     (location, key, value) ->
                         writeStorageTrieNode(
                             bonsaiUpdater, updatedAddressHash, location, key, value)));
-        final Hash newStorageRoot = Hash.wrap(storageTrie.getRootHash());
+        // only use storage root of the trie when trie is enabled
-        accountUpdated.setStorageRoot(newStorageRoot);
+        if (!worldStateConfig.isTrieDisabled()) {
+          final Hash newStorageRoot = Hash.wrap(storageTrie.getRootHash());
+          accountUpdated.setStorageRoot(newStorageRoot);
+        }
       }
     }
     // for manicured tries and composting, trim and compost here
@@ -347,13 +365,6 @@ public class BonsaiWorldState extends DiffBasedWorldState {
         accumulator.copy());
   }
 
-  @Override
-  public MutableWorldState freeze() {
-    this.isFrozen = true;
-    this.worldStateKeyValueStorage = new BonsaiWorldStateLayerStorage(getWorldStateStorage());
-    return this;
-  }
-
   @Override
   public Account get(final Address address) {
     return getWorldStateStorage()
@@ -362,11 +373,6 @@ public class BonsaiWorldState extends DiffBasedWorldState {
         .orElse(null);
   }
 
-  @Override
-  public Optional<Bytes> getCode(@Nonnull final Address address, final Hash codeHash) {
-    return getWorldStateStorage().getCode(codeHash, address.addressHash());
-  }
-
   protected Optional<Bytes> getAccountStateTrieNode(final Bytes location, final Bytes32 nodeHash) {
     return getWorldStateStorage().getAccountStateTrieNode(location, nodeHash);
   }
@@ -423,16 +429,26 @@ public class BonsaiWorldState extends DiffBasedWorldState {
 
   @Override
   public Map<Bytes32, Bytes> getAllAccountStorage(final Address address, final Hash rootHash) {
-    final StoredMerklePatriciaTrie<Bytes, Bytes> storageTrie =
+    final MerkleTrie<Bytes, Bytes> storageTrie =
         createTrie(
             (location, key) -> getStorageTrieNode(address.addressHash(), location, key), rootHash);
     return storageTrie.entriesFrom(Bytes32.ZERO, Integer.MAX_VALUE);
   }
 
-  private StoredMerklePatriciaTrie<Bytes, Bytes> createTrie(
+  @Override
-      final NodeLoader nodeLoader, final Bytes32 rootHash) {
+  public MutableWorldState freeze() {
-    return new StoredMerklePatriciaTrie<>(
+    this.worldStateConfig.setFrozen(true);
-        nodeLoader, rootHash, Function.identity(), Function.identity());
+    this.worldStateKeyValueStorage = new BonsaiWorldStateLayerStorage(getWorldStateStorage());
+    return this;
+  }
+
+  private MerkleTrie<Bytes, Bytes> createTrie(final NodeLoader nodeLoader, final Bytes32 rootHash) {
+    if (worldStateConfig.isTrieDisabled()) {
+      return new NoOpMerkleTrie<>();
+    } else {
+      return new StoredMerklePatriciaTrie<>(
+          nodeLoader, rootHash, Function.identity(), Function.identity());
+    }
   }
 
   protected Hash hashAndSavePreImage(final Bytes value) {

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/DiffBasedWorldStateProvider.java

@@ -26,6 +26,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.common.cache.DiffBasedCached
 import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogManager;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.accumulator.DiffBasedWorldStateUpdateAccumulator;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
@@ -54,6 +55,7 @@ public abstract class DiffBasedWorldStateProvider implements WorldStateArchive {
   protected DiffBasedWorldState persistedState;
 
   protected final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage;
+  protected final DiffBasedWorldStateConfig defaultWorldStateConfig;
 
   public DiffBasedWorldStateProvider(
       final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage,
@@ -70,6 +72,7 @@ public abstract class DiffBasedWorldStateProvider implements WorldStateArchive {
             maxLayersToLoad.orElse(DiffBasedCachedWorldStorageManager.RETAINED_LAYERS),
             pluginContext);
     this.blockchain = blockchain;
+    this.defaultWorldStateConfig = new DiffBasedWorldStateConfig();
   }
 
   public DiffBasedWorldStateProvider(
@@ -81,6 +84,7 @@ public abstract class DiffBasedWorldStateProvider implements WorldStateArchive {
     // TODO: de-dup constructors
     this.trieLogManager = trieLogManager;
     this.blockchain = blockchain;
+    this.defaultWorldStateConfig = new DiffBasedWorldStateConfig();
   }
 
   protected void provideCachedWorldStorageManager(
@@ -252,6 +256,19 @@ public abstract class DiffBasedWorldStateProvider implements WorldStateArchive {
     return persistedState;
   }
 
+  public DiffBasedWorldStateConfig getDefaultWorldStateConfig() {
+    return defaultWorldStateConfig;
+  }
+
+  public void disableTrie() {
+    defaultWorldStateConfig.setTrieDisabled(true);
+    worldStateKeyValueStorage.clearTrie();
+  }
+
+  public DiffBasedWorldStateKeyValueStorage getWorldStateKeyValueStorage() {
+    return worldStateKeyValueStorage;
+  }
+
   public TrieLogManager getTrieLogManager() {
     return trieLogManager;
   }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/StorageSubscriber.java

@@ -21,5 +21,7 @@ public interface StorageSubscriber {
 
   default void onClearTrieLog() {}
 
+  default void onClearTrie() {}
+
   default void onCloseStorage() {}
 }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/cache/DiffBasedCachedWorldStorageManager.java

@@ -23,6 +23,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.common.StorageSubscriber;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedLayeredWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.storage.DiffBasedWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 
 import java.util.ArrayList;
@@ -33,6 +34,7 @@ import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -46,6 +48,7 @@ public abstract class DiffBasedCachedWorldStorageManager implements StorageSubsc
       LoggerFactory.getLogger(DiffBasedCachedWorldStorageManager.class);
   private final DiffBasedWorldStateProvider archive;
   private final EvmConfiguration evmConfiguration;
+  protected final Supplier<DiffBasedWorldStateConfig> defaultBonsaiWorldStateConfigSupplier;
   private final Cache<Hash, BlockHeader> stateRootToBlockHeaderCache =
       Caffeine.newBuilder()
           .maximumSize(RETAINED_LAYERS)
@@ -59,18 +62,26 @@ public abstract class DiffBasedCachedWorldStorageManager implements StorageSubsc
       final DiffBasedWorldStateProvider archive,
       final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage,
       final Map<Bytes32, DiffBasedCachedWorldView> cachedWorldStatesByHash,
-      final EvmConfiguration evmConfiguration) {
+      final EvmConfiguration evmConfiguration,
+      final Supplier<DiffBasedWorldStateConfig> defaultBonsaiWorldStateConfigSupplier) {
     worldStateKeyValueStorage.subscribe(this);
     this.rootWorldStateStorage = worldStateKeyValueStorage;
     this.cachedWorldStatesByHash = cachedWorldStatesByHash;
     this.archive = archive;
     this.evmConfiguration = evmConfiguration;
+    this.defaultBonsaiWorldStateConfigSupplier = defaultBonsaiWorldStateConfigSupplier;
   }
 
   public DiffBasedCachedWorldStorageManager(
       final DiffBasedWorldStateProvider archive,
-      final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage) {
+      final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage,
-    this(archive, worldStateKeyValueStorage, new ConcurrentHashMap<>(), EvmConfiguration.DEFAULT);
+      final Supplier<DiffBasedWorldStateConfig> defaultBonsaiWorldStateConfigSupplier) {
+    this(
+        archive,
+        worldStateKeyValueStorage,
+        new ConcurrentHashMap<>(),
+        EvmConfiguration.DEFAULT,
+        defaultBonsaiWorldStateConfigSupplier);
   }
 
   public synchronized void addCachedLayer(
@@ -263,6 +274,11 @@ public abstract class DiffBasedCachedWorldStorageManager implements StorageSubsc
     this.cachedWorldStatesByHash.clear();
   }
 
+  @Override
+  public void onClearTrie() {
+    this.cachedWorldStatesByHash.clear();
+  }
+
   @Override
   public void onCloseStorage() {
     this.cachedWorldStatesByHash.clear();

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/storage/DiffBasedWorldStateKeyValueStorage.java

@@ -174,6 +174,11 @@ public abstract class DiffBasedWorldStateKeyValueStorage
     trieLogStorage.clear();
   }
 
+  public void clearTrie() {
+    subscribers.forEach(StorageSubscriber::onClearTrie);
+    composedWorldStateStorage.clear(TRIE_BRANCH_STORAGE);
+  }
+
   public void clearFlatDatabase() {
     subscribers.forEach(StorageSubscriber::onClearFlatDatabaseStorage);
     getFlatDbStrategy().resetOnResync(composedWorldStateStorage);

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/trielog/TrieLogManager.java

@@ -34,6 +34,7 @@ import java.util.Optional;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
+import org.apache.tuweni.bytes.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -147,15 +148,46 @@ public class TrieLogManager {
       trieLogService.getObservers().forEach(trieLogObservers::subscribe);
 
       // return the TrieLogFactory implementation from the TrieLogService
-      return trieLogService.getTrieLogFactory();
+      if (trieLogService.getTrieLogFactory().isPresent()) {
-    } else {
+        return trieLogService.getTrieLogFactory().get();
-      // Otherwise default to TrieLogFactoryImpl
+      }
-      return new TrieLogFactoryImpl();
     }
+    // Otherwise default to TrieLogFactoryImpl
+    return new TrieLogFactoryImpl();
   }
 
   private TrieLogProvider getTrieLogProvider() {
     return new TrieLogProvider() {
+      @Override
+      public Optional<Bytes> getRawTrieLogLayer(final Hash blockHash) {
+        return rootWorldStateStorage.getTrieLog(blockHash).map(Bytes::wrap);
+      }
+
+      @Override
+      public Optional<Bytes> getRawTrieLogLayer(final long blockNumber) {
+        return TrieLogManager.this
+            .blockchain
+            .getBlockHeader(blockNumber)
+            .map(BlockHeader::getHash)
+            .flatMap(this::getRawTrieLogLayer);
+      }
+
+      @Override
+      public void saveRawTrieLogLayer(
+          final Hash blockHash, final long blockNumber, final Bytes trieLog) {
+        final DiffBasedWorldStateKeyValueStorage.Updater updater = rootWorldStateStorage.updater();
+        updater
+            .getTrieLogStorageTransaction()
+            .put(blockHash.toArrayUnsafe(), trieLog.toArrayUnsafe());
+        updater.commit();
+        // TODO maybe find a way to have a clean and complete trielog for observers
+        trieLogObservers.forEach(
+            o ->
+                o.onTrieLogAdded(
+                    new TrieLogAddedEvent(
+                        new TrieLogLayer().setBlockHash(blockHash).setBlockNumber(blockNumber))));
+      }
+
       @Override
       public Optional<TrieLog> getTrieLogLayer(final Hash blockHash) {
         return TrieLogManager.this.getTrieLogLayer(blockHash);

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/worldview/DiffBasedWorldState.java

@@ -60,12 +60,13 @@ public abstract class DiffBasedWorldState
 
   protected Hash worldStateRootHash;
   protected Hash worldStateBlockHash;
-  protected boolean isFrozen;
+  protected DiffBasedWorldStateConfig worldStateConfig;
 
   protected DiffBasedWorldState(
       final DiffBasedWorldStateKeyValueStorage worldStateKeyValueStorage,
       final DiffBasedCachedWorldStorageManager cachedWorldStorageManager,
-      final TrieLogManager trieLogManager) {
+      final TrieLogManager trieLogManager,
+      final DiffBasedWorldStateConfig diffBasedWorldStateConfig) {
     this.worldStateKeyValueStorage = worldStateKeyValueStorage;
     this.worldStateRootHash =
         Hash.wrap(
@@ -76,11 +77,12 @@ public abstract class DiffBasedWorldState
             Bytes32.wrap(worldStateKeyValueStorage.getWorldStateBlockHash().orElse(Hash.ZERO)));
     this.cachedWorldStorageManager = cachedWorldStorageManager;
     this.trieLogManager = trieLogManager;
+    this.worldStateConfig = diffBasedWorldStateConfig;
   }
 
   /**
    * Having a protected method to override the accumulator solves the chicken-egg problem of needing
-   * a worldstate reference (this) when construction the Accumulator.
+   * a worldstate reference (this) when constructing the Accumulator.
    *
    * @param accumulator accumulator to use.
    */
@@ -134,6 +136,15 @@ public abstract class DiffBasedWorldState
     return accumulator;
   }
 
+  protected Hash unsafeRootHashUpdate(
+      final BlockHeader blockHeader,
+      final DiffBasedWorldStateKeyValueStorage.Updater stateUpdater) {
+    // calling calculateRootHash in order to update the state
+    calculateRootHash(
+        worldStateConfig.isFrozen() ? Optional.empty() : Optional.of(stateUpdater), accumulator);
+    return blockHeader.getStateRoot();
+  }
+
   @Override
   public void persist(final BlockHeader blockHeader) {
     final Optional<BlockHeader> maybeBlockHeader = Optional.ofNullable(blockHeader);
@@ -151,19 +162,32 @@ public abstract class DiffBasedWorldState
     Runnable saveTrieLog = () -> {};
 
     try {
-      final Hash newWorldStateRootHash =
+      final Hash calculatedRootHash;
-          calculateRootHash(isFrozen ? Optional.empty() : Optional.of(stateUpdater), accumulator);
+
+      if (blockHeader == null || !worldStateConfig.isTrieDisabled()) {
+        calculatedRootHash =
+            calculateRootHash(
+                worldStateConfig.isFrozen() ? Optional.empty() : Optional.of(stateUpdater),
+                accumulator);
+      } else {
+        // if the trie is disabled, we cannot calculate the state root, so we directly use the root
+        // of the block. It's important to understand that in all networks,
+        // the state root must be validated independently and the block should not be trusted
+        // implicitly. This mode
+        // can be used in cases where Besu would just be a follower of another trusted client.
+        calculatedRootHash = unsafeRootHashUpdate(blockHeader, stateUpdater);
+      }
       // if we are persisted with a block header, and the prior state is the parent
       // then persist the TrieLog for that transition.
       // If specified but not a direct descendant simply store the new block hash.
       if (blockHeader != null) {
-        verifyWorldStateRoot(newWorldStateRootHash, blockHeader);
+        verifyWorldStateRoot(calculatedRootHash, blockHeader);
         saveTrieLog =
             () -> {
-              trieLogManager.saveTrieLog(localCopy, newWorldStateRootHash, blockHeader, this);
+              trieLogManager.saveTrieLog(localCopy, calculatedRootHash, blockHeader, this);
               // not save a frozen state in the cache
-              if (!isFrozen) {
+              if (!worldStateConfig.isFrozen()) {
-                cachedWorldStorageManager.addCachedLayer(blockHeader, newWorldStateRootHash, this);
+                cachedWorldStorageManager.addCachedLayer(blockHeader, calculatedRootHash, this);
               }
             };
 
@@ -178,8 +202,8 @@ public abstract class DiffBasedWorldState
 
       stateUpdater
           .getWorldStateTransaction()
-          .put(TRIE_BRANCH_STORAGE, WORLD_ROOT_HASH_KEY, newWorldStateRootHash.toArrayUnsafe());
+          .put(TRIE_BRANCH_STORAGE, WORLD_ROOT_HASH_KEY, calculatedRootHash.toArrayUnsafe());
-      worldStateRootHash = newWorldStateRootHash;
+      worldStateRootHash = calculatedRootHash;
       success = true;
     } finally {
       if (success) {
@@ -194,7 +218,7 @@ public abstract class DiffBasedWorldState
   }
 
   protected void verifyWorldStateRoot(final Hash calculatedStateRoot, final BlockHeader header) {
-    if (!calculatedStateRoot.equals(header.getStateRoot())) {
+    if (!worldStateConfig.isTrieDisabled() && !calculatedStateRoot.equals(header.getStateRoot())) {
       throw new RuntimeException(
           "World State Root does not match expected value, header "
               + header.getStateRoot().toHexString()
@@ -210,7 +234,7 @@ public abstract class DiffBasedWorldState
 
   @Override
   public Hash rootHash() {
-    if (isFrozen && accumulator.isAccumulatorStateChanged()) {
+    if (worldStateConfig.isFrozen() && accumulator.isAccumulatorStateChanged()) {
       worldStateRootHash = calculateRootHash(Optional.empty(), accumulator.copy());
       accumulator.resetAccumulatorStateChanged();
     }
@@ -285,7 +309,7 @@ public abstract class DiffBasedWorldState
     try {
       if (!isPersisted()) {
         this.worldStateKeyValueStorage.close();
-        if (isFrozen) {
+        if (worldStateConfig.isFrozen()) {
           closeFrozenStorage();
         }
       }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/worldview/DiffBasedWorldStateConfig.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.trie.diffbased.common.worldview;
+
+public class DiffBasedWorldStateConfig {
+
+  private boolean isFrozen;
+
+  private boolean isTrieDisabled;
+
+  public DiffBasedWorldStateConfig() {
+    this(false, false);
+  }
+
+  public DiffBasedWorldStateConfig(final boolean isTrieDisabled) {
+    this(false, isTrieDisabled);
+  }
+
+  public DiffBasedWorldStateConfig(final DiffBasedWorldStateConfig config) {
+    this(config.isFrozen(), config.isTrieDisabled());
+  }
+
+  public DiffBasedWorldStateConfig(final boolean isFrozen, final boolean isTrieDisabled) {
+    this.isFrozen = isFrozen;
+    this.isTrieDisabled = isTrieDisabled;
+  }
+
+  /**
+   * Checks if the world state is frozen. When the world state is frozen, it cannot mutate.
+   *
+   * @return true if the world state is frozen, false otherwise.
+   */
+  public boolean isFrozen() {
+    return isFrozen;
+  }
+
+  /**
+   * Sets the frozen status of the world state. When the world state is frozen, it cannot mutate.
+   *
+   * @param frozen the new frozen status to set.
+   */
+  public void setFrozen(final boolean frozen) {
+    isFrozen = frozen;
+  }
+
+  /**
+   * Checks if the trie is disabled for the world state. When the trie is disabled, the world state
+   * will only work with the flat database and not the trie. In this mode, it's impossible to verify
+   * the state root.
+   *
+   * @return true if the trie is disabled, false otherwise.
+   */
+  public boolean isTrieDisabled() {
+    return isTrieDisabled;
+  }
+
+  /**
+   * Sets the disabled status of the trie for the world state. When the trie is disabled, the world
+   * state will only work with the flat database and not the trie. In this mode, it's impossible to
+   * verify the state root.
+   *
+   * @param trieDisabled the new disabled status to set for the trie.
+   */
+  public void setTrieDisabled(final boolean trieDisabled) {
+    isTrieDisabled = trieDisabled;
+  }
+}

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/BlockImportExceptionHandlingTest.java

@@ -46,6 +46,7 @@ import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.BonsaiWorldStateProvider;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
@@ -100,7 +101,8 @@ class BlockImportExceptionHandlingTest {
               (BonsaiWorldStateProvider) worldStateArchive,
               (BonsaiWorldStateKeyValueStorage)
                   worldStateStorageCoordinator.worldStateKeyValueStorage(),
-              EvmConfiguration.DEFAULT));
+              EvmConfiguration.DEFAULT,
+              new DiffBasedWorldStateConfig()));
 
   private final BadBlockManager badBlockManager = new BadBlockManager();
 

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/chain/ChainDataPrunerTest.java

@@ -48,6 +48,7 @@ public class ChainDataPrunerTest {
             new ChainDataPrunerStorage(new InMemoryKeyValueStorage()),
             512,
             0,
+            // completed
             new BlockingExecutor());
     Block genesisBlock = gen.genesisBlock();
     final MutableBlockchain blockchain =
@@ -87,6 +88,7 @@ public class ChainDataPrunerTest {
             new ChainDataPrunerStorage(new InMemoryKeyValueStorage()),
             512,
             0,
+            // completed
             new BlockingExecutor());
     Block genesisBlock = gen.genesisBlock();
     final MutableBlockchain blockchain =

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

@@ -199,6 +199,16 @@ public abstract class AbstractIsolationTests {
         .withCommonConfiguration(
             new BesuConfiguration() {
 
+              @Override
+              public Optional<String> getRpcHttpHost() {
+                return Optional.empty();
+              }
+
+              @Override
+              public Optional<Integer> getRpcHttpPort() {
+                return Optional.empty();
+              }
+
               @Override
               public Path getStoragePath() {
                 return tempData.resolve("database");

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/LogRollingTests.java

@@ -32,6 +32,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.trielog.TrieLogFactor
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldState;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldStateUpdateAccumulator;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogLayer;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
 import org.hyperledger.besu.evm.account.MutableAccount;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
@@ -162,7 +163,8 @@ class LogRollingTests {
             archive,
             new BonsaiWorldStateKeyValueStorage(
                 provider, new NoOpMetricsSystem(), DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
     final WorldUpdater updater = worldState.updater();
 
     final MutableAccount mutableAccount = updater.createAccount(addressOne, 1, Wei.of(1L));
@@ -178,7 +180,8 @@ class LogRollingTests {
                 secondProvider,
                 new NoOpMetricsSystem(),
                 DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
     final BonsaiWorldStateUpdateAccumulator secondUpdater =
         (BonsaiWorldStateUpdateAccumulator) secondWorldState.updater();
 
@@ -210,7 +213,8 @@ class LogRollingTests {
             archive,
             new BonsaiWorldStateKeyValueStorage(
                 provider, new NoOpMetricsSystem(), DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
 
     final WorldUpdater updater = worldState.updater();
     final MutableAccount mutableAccount = updater.createAccount(addressOne, 1, Wei.of(1L));
@@ -234,7 +238,8 @@ class LogRollingTests {
                 secondProvider,
                 new NoOpMetricsSystem(),
                 DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
     final BonsaiWorldStateUpdateAccumulator secondUpdater =
         (BonsaiWorldStateUpdateAccumulator) secondWorldState.updater();
 
@@ -267,7 +272,8 @@ class LogRollingTests {
             archive,
             new BonsaiWorldStateKeyValueStorage(
                 provider, new NoOpMetricsSystem(), DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
 
     final WorldUpdater updater = worldState.updater();
     final MutableAccount mutableAccount = updater.createAccount(addressOne, 1, Wei.of(1L));
@@ -298,7 +304,8 @@ class LogRollingTests {
                 secondProvider,
                 new NoOpMetricsSystem(),
                 DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
 
     final WorldUpdater secondUpdater = secondWorldState.updater();
     final MutableAccount secondMutableAccount =

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/RollingImport.java

@@ -28,6 +28,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.trielog.TrieLogFactor
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldState;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.worldview.BonsaiWorldStateUpdateAccumulator;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogLayer;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
@@ -57,7 +58,8 @@ public class RollingImport {
             archive,
             new BonsaiWorldStateKeyValueStorage(
                 provider, new NoOpMetricsSystem(), DataStorageConfiguration.DEFAULT_BONSAI_CONFIG),
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
     final SegmentedInMemoryKeyValueStorage worldStateKeyValueStorage =
         (SegmentedInMemoryKeyValueStorage)
             provider.getStorageBySegmentIdentifiers(

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/trie/diffbased/bonsai/worldview/BonsaiWorldStateTest.java

@@ -24,6 +24,7 @@ import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.DiffBasedValue;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 
 import java.util.HashMap;
@@ -61,7 +62,8 @@ class BonsaiWorldStateTest {
         new BonsaiWorldState(
             InMemoryKeyValueStorageProvider.createBonsaiInMemoryWorldStateArchive(blockchain),
             bonsaiWorldStateKeyValueStorage,
-            EvmConfiguration.DEFAULT);
+            EvmConfiguration.DEFAULT,
+            new DiffBasedWorldStateConfig());
   }
 
   @ParameterizedTest

ethereum/eth/src/jmh/java/org/hyperledger/besu/ethereum/eth/sync/worldstate/WorldStateDownloaderBenchmark.java

@@ -24,7 +24,6 @@ import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
-import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.MutableWorldState;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
@@ -165,8 +164,7 @@ public class WorldStateDownloaderBenchmark {
 
   private StorageProvider createKeyValueStorageProvider(final Path dataDir, final Path dbDir) {
     final var besuConfiguration = new BesuConfigurationImpl();
-    besuConfiguration.init(
+    besuConfiguration.init(dataDir, dbDir, DataStorageConfiguration.DEFAULT_CONFIG);
-        dataDir, dbDir, DataStorageConfiguration.DEFAULT_CONFIG, MiningParameters.newDefault());
     return new KeyValueStorageProviderBuilder()
         .withStorageFactory(
             new RocksDBKeyValueStorageFactory(

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/AbstractSyncTargetManager.java

@@ -28,6 +28,7 @@ import org.hyperledger.besu.plugin.services.MetricsSystem;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,6 +37,8 @@ public abstract class AbstractSyncTargetManager {
 
   private static final Logger LOG = LoggerFactory.getLogger(AbstractSyncTargetManager.class);
 
+  private final AtomicBoolean cancelled = new AtomicBoolean(false);
+
   private final SynchronizerConfiguration config;
   private final ProtocolSchedule protocolSchedule;
   private final ProtocolContext protocolContext;
@@ -56,6 +59,9 @@ public abstract class AbstractSyncTargetManager {
   }
 
   public CompletableFuture<SyncTarget> findSyncTarget() {
+    if (isCancelled()) {
+      return completedFuture(null);
+    }
     return selectBestAvailableSyncTarget()
         .thenCompose(
             maybeBestPeer -> {
@@ -99,6 +105,10 @@ public abstract class AbstractSyncTargetManager {
             });
   }
 
+  public synchronized void cancel() {
+    cancelled.set(true);
+  }
+
   protected Optional<SyncTarget> finalizeSelectedSyncTarget(final SyncTarget syncTarget) {
     return Optional.of(syncTarget);
   }
@@ -115,5 +125,9 @@ public abstract class AbstractSyncTargetManager {
         .timeout(WaitForPeerTask.create(ethContext, metricsSystem), Duration.ofSeconds(5));
   }
 
+  private boolean isCancelled() {
+    return cancelled.get();
+  }
+
   public abstract boolean shouldContinueDownloading();
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/PipelineChainDownloader.java

@@ -85,6 +85,7 @@ public class PipelineChainDownloader implements ChainDownloader {
   @Override
   public synchronized void cancel() {
     cancelled.set(true);
+    syncTargetManager.cancel();
     if (currentDownloadPipeline != null) {
       currentDownloadPipeline.abort();
     }

ethereum/evmtool/src/main/java/org/hyperledger/besu/evmtool/EvmToolCommandOptionsModule.java

@@ -87,7 +87,7 @@ public class EvmToolCommandOptionsModule {
   @Singleton
   BesuConfiguration provideBesuConfiguration() {
     final var besuConfiguration = new BesuConfigurationImpl();
-    besuConfiguration.init(dataPath, dataPath.resolve(BesuController.DATABASE_PATH), null, null);
+    besuConfiguration.init(dataPath, dataPath.resolve(BesuController.DATABASE_PATH), null);
     return besuConfiguration;
   }
 

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/discovery/internal/PeerDiscoveryController.java

@@ -260,6 +260,7 @@ public class PeerDiscoveryController {
               l.clear();
             });
     inflightInteractions.clear();
+    recursivePeerRefreshState.cancel();
     return CompletableFuture.completedFuture(null);
   }
 

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/discovery/internal/RecursivePeerRefreshState.java

@@ -105,6 +105,10 @@ public class RecursivePeerRefreshState {
   }
 
   private void bondingInitiateRound() {
+    if (!iterativeSearchInProgress) {
+      // cancelled so we can ignore
+      return;
+    }
     currentRoundTimeout.ifPresent(RoundTimeout::cancelTimeout);
     final List<MetadataPeer> candidates = bondingRoundCandidates();
     if (candidates.isEmpty()) {
@@ -137,6 +141,10 @@ public class RecursivePeerRefreshState {
   }
 
   private void bondingCancelOutstandingRequests() {
+    if (!iterativeSearchInProgress) {
+      // cancelled so we can ignore
+      return;
+    }
     LOG.debug("Bonding round timed out");
     for (final Map.Entry<Bytes, MetadataPeer> entry : oneTrueMap.entrySet()) {
       final MetadataPeer metadataPeer = entry.getValue();
@@ -149,6 +157,10 @@ public class RecursivePeerRefreshState {
   }
 
   private void neighboursInitiateRound() {
+    if (!iterativeSearchInProgress) {
+      // cancelled so we can ignore
+      return;
+    }
     currentRoundTimeout.ifPresent(RoundTimeout::cancelTimeout);
     final List<MetadataPeer> candidates = neighboursRoundCandidates();
     if (candidates.isEmpty() || reachedMaximumNumberOfRounds()) {
@@ -172,6 +184,10 @@ public class RecursivePeerRefreshState {
   }
 
   private void neighboursCancelOutstandingRequests() {
+    if (!iterativeSearchInProgress) {
+      // cancelled so we can ignore
+      return;
+    }
     LOG.debug("Neighbours round timed out");
     for (final Map.Entry<Bytes, MetadataPeer> entry : oneTrueMap.entrySet()) {
       final MetadataPeer metadataPeer = entry.getValue();
@@ -218,6 +234,10 @@ public class RecursivePeerRefreshState {
   }
 
   void onBondingComplete(final DiscoveryPeer peer) {
+    if (!iterativeSearchInProgress) {
+      // cancelled so we can ignore
+      return;
+    }
     final MetadataPeer iterationParticipant = oneTrueMap.get(peer.getId());
     if (iterationParticipant == null) {
       return;

ethereum/referencetests/src/main/java/org/hyperledger/besu/ethereum/referencetests/BonsaiReferenceTestWorldState.java

@@ -29,6 +29,7 @@ import org.hyperledger.besu.ethereum.trie.diffbased.common.cache.DiffBasedCached
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogAddedEvent;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.trielog.TrieLogManager;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldState;
+import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.DiffBasedWorldStateConfig;
 import org.hyperledger.besu.ethereum.trie.diffbased.common.worldview.accumulator.DiffBasedWorldStateUpdateAccumulator;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
@@ -66,7 +67,8 @@ public class BonsaiReferenceTestWorldState extends BonsaiWorldState
         bonsaiCachedMerkleTrieLoader,
         cachedWorldStorageManager,
         trieLogManager,
-        evmConfiguration);
+        evmConfiguration,
+        new DiffBasedWorldStateConfig());
     this.refTestStorage = worldStateKeyValueStorage;
     this.preImageProxy = preImageProxy;
     this.evmConfiguration = evmConfiguration;
@@ -194,7 +196,8 @@ public class BonsaiReferenceTestWorldState extends BonsaiWorldState
             bonsaiCachedMerkleTrieLoader,
             cachedWorldStorageManager,
             trieLogManager,
-            evmConfiguration);
+            evmConfiguration,
+            new DiffBasedWorldStateConfig());
     if (isFrozen) {
       bonsaiWorldState.freeze(); // freeze state
     }

ethereum/trie/src/main/java/org/hyperledger/besu/ethereum/trie/NoOpMerkleTrie.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.trie;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.tuweni.bytes.Bytes;
+import org.apache.tuweni.bytes.Bytes32;
+
+/**
+ * A noop {@link MerkleTrie}.
+ *
+ * @param <V> The type of values of this trie.
+ */
+public class NoOpMerkleTrie<K extends Bytes, V> implements MerkleTrie<K, V> {
+
+  public NoOpMerkleTrie() {}
+
+  @Override
+  public Optional<V> get(final K key) {
+    return Optional.empty();
+  }
+
+  @Override
+  public Optional<V> getPath(final K path) {
+    return Optional.empty();
+  }
+
+  @Override
+  public Proof<V> getValueWithProof(final K key) {
+    return new Proof<>(Optional.empty(), new ArrayList<>());
+  }
+
+  @Override
+  public void put(final K key, final V value) {
+    // noop
+  }
+
+  @Override
+  public void putPath(final K path, final V value) {
+    // noop
+  }
+
+  @Override
+  public void put(final K key, final PathNodeVisitor<V> putVisitor) {
+    // noop
+  }
+
+  @Override
+  public void remove(final K key) {
+    // noop
+  }
+
+  @Override
+  public void removePath(final K path, final PathNodeVisitor<V> removeVisitor) {
+    // noop
+  }
+
+  @Override
+  public Bytes32 getRootHash() {
+    return EMPTY_TRIE_NODE_HASH;
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + "[" + getRootHash() + "]";
+  }
+
+  @Override
+  public void commit(final NodeUpdater nodeUpdater) {
+    // Nothing to do here
+  }
+
+  @Override
+  public void commit(final NodeUpdater nodeUpdater, final CommitVisitor<V> commitVisitor) {
+    // Nothing to do here
+  }
+
+  @Override
+  public Map<Bytes32, V> entriesFrom(final Bytes32 startKeyHash, final int limit) {
+    return new HashMap<>();
+  }
+
+  @Override
+  public Map<Bytes32, V> entriesFrom(final Function<Node<V>, Map<Bytes32, V>> handler) {
+    return new HashMap<>();
+  }
+
+  @Override
+  public void visitAll(final Consumer<Node<V>> nodeConsumer) {
+    // noop
+  }
+
+  @Override
+  public CompletableFuture<Void> visitAll(
+      final Consumer<Node<V>> nodeConsumer, final ExecutorService executorService) {
+    return CompletableFuture.completedFuture(null);
+  }
+
+  @Override
+  public void visitLeafs(final TrieIterator.LeafHandler<V> handler) {
+    // nopop
+  }
+}

plugin-api/build.gradle

@@ -70,7 +70,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 = 'xU0LMvStiFihBKBxkfNHB7oIg0PUHWkPuv2yt1GVC94='
+  knownHash = 'wMhttXj2aWFgpN9msxHz/FwQVovpYRxNysnZEpwZYxg='
 }
 check.dependsOn('checkAPIChanges')
 

plugin-api/src/main/java/org/hyperledger/besu/plugin/BesuPlugin.java

@@ -63,6 +63,9 @@ public interface BesuPlugin {
    */
   void start();
 
+  /** Hook to execute plugin setup code after external services */
+  default void afterExternalServicePostMainLoop() {}
+
   /**
    * Called when the plugin is being reloaded. This method will be called through a dedicated JSON
    * RPC endpoint. If not overridden this method does nothing for convenience. The plugin should

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/BesuConfiguration.java

@@ -20,10 +20,25 @@ import org.hyperledger.besu.plugin.services.storage.DataStorageConfiguration;
 import org.hyperledger.besu.plugin.services.storage.DataStorageFormat;
 
 import java.nio.file.Path;
+import java.util.Optional;
 
 /** Generally useful configuration provided by Besu. */
 public interface BesuConfiguration extends BesuService {
 
+  /**
+   * Get the configured RPC http host.
+   *
+   * @return the configured RPC http host.
+   */
+  Optional<String> getRpcHttpHost();
+
+  /**
+   * Get the configured RPC http port.
+   *
+   * @return the configured RPC http port.
+   */
+  Optional<Integer> getRpcHttpPort();
+
   /**
    * Location of the working directory of the storage in the file system running the client.
    *

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/BesuEvents.java

@@ -93,6 +93,14 @@ public interface BesuEvents extends BesuService {
    */
   void removeBlockReorgListener(long listenerIdentifier);
 
+  /**
+   * Add an initial sync completion listener.
+   *
+   * @param listener to subscribe to initial sync completion events
+   * @return id of listener subscription
+   */
+  long addInitialSyncCompletionListener(final InitialSyncCompletionListener listener);
+
   /**
    * Add a listener watching new transactions added to the node.
    *

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/BlockchainService.java

@@ -17,9 +17,12 @@ package org.hyperledger.besu.plugin.services;
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.plugin.Unstable;
+import org.hyperledger.besu.plugin.data.BlockBody;
 import org.hyperledger.besu.plugin.data.BlockContext;
 import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.TransactionReceipt;
 
+import java.util.List;
 import java.util.Optional;
 
 /** A service that plugins can use to query blocks by number */
@@ -40,6 +43,23 @@ public interface BlockchainService extends BesuService {
    */
   Hash getChainHeadHash();
 
+  /**
+   * Get the receipts for a block by block hash
+   *
+   * @param blockHash the block hash
+   * @return the transaction receipts
+   */
+  Optional<List<TransactionReceipt>> getReceiptsByBlockHash(Hash blockHash);
+
+  /**
+   * Store a block
+   *
+   * @param blockHeader the block header
+   * @param blockBody the block body
+   * @param receipts the transaction receipts
+   */
+  void storeBlock(BlockHeader blockHeader, BlockBody blockBody, List<TransactionReceipt> receipts);
+
   /**
    * Get the block header of the chain head
    *
@@ -53,4 +73,18 @@ public interface BlockchainService extends BesuService {
    * @return base fee of the next block or empty if the fee market does not support base fee
    */
   Optional<Wei> getNextBlockBaseFee();
+
+  /**
+   * Get the block hash of the safe block
+   *
+   * @return the block hash of the safe block
+   */
+  Optional<Hash> getSafeBlock();
+
+  /**
+   * Get the block hash of the finalized block
+   *
+   * @return the block hash of the finalized block
+   */
+  Optional<Hash> getFinalizedBlock();
 }

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/TrieLogService.java

@@ -19,6 +19,7 @@ import org.hyperledger.besu.plugin.services.trielogs.TrieLogFactory;
 import org.hyperledger.besu.plugin.services.trielogs.TrieLogProvider;
 
 import java.util.List;
+import java.util.Optional;
 
 /**
  * A service interface for registering observers for trie log events.
@@ -39,7 +40,7 @@ public interface TrieLogService extends BesuService {
    *
    * @return the TrieLogFactory implementation
    */
-  TrieLogFactory getTrieLogFactory();
+  Optional<TrieLogFactory> getTrieLogFactory();
 
   /**
    * Configure a TrieLogProvider implementation to use for retrieving stored TrieLogs.

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/p2p/P2PService.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.p2p;
+
+import org.hyperledger.besu.plugin.services.BesuService;
+
+/** Service to enable and disable P2P service. */
+public interface P2PService extends BesuService {
+
+  /** Enables P2P discovery. */
+  void enableDiscovery();
+
+  /** Disables P2P discovery. */
+  void disableDiscovery();
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/rlp/RlpConverterService.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.rlp;
+
+import org.hyperledger.besu.plugin.data.BlockBody;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.TransactionReceipt;
+import org.hyperledger.besu.plugin.services.BesuService;
+
+import org.apache.tuweni.bytes.Bytes;
+
+/** RLP Serialiaztion/Deserialization service. */
+public interface RlpConverterService extends BesuService {
+
+  /**
+   * Builds a block header from RLP.
+   *
+   * @param rlp the RLP to build the block header from.
+   * @return the block header.
+   */
+  BlockHeader buildHeaderFromRlp(final Bytes rlp);
+
+  /**
+   * Builds a block body from RLP.
+   *
+   * @param rlp the RLP to build the block body from.
+   * @return the block body.
+   */
+  BlockBody buildBodyFromRlp(final Bytes rlp);
+
+  /**
+   * Builds a transaction receipt from RLP.
+   *
+   * @param rlp the RLP to build the transaction receipt from.
+   * @return the transaction receipt.
+   */
+  TransactionReceipt buildReceiptFromRlp(final Bytes rlp);
+
+  /**
+   * RLP encodes a block header.
+   *
+   * @param blockHeader the block header to build RLP from.
+   * @return the RLP.
+   */
+  Bytes buildRlpFromHeader(final BlockHeader blockHeader);
+
+  /**
+   * RLP encodes a block body.
+   *
+   * @param blockBody the block body to build RLP from.
+   * @return the RLP.
+   */
+  Bytes buildRlpFromBody(final BlockBody blockBody);
+
+  /**
+   * RLP encodes a transaction receipt.
+   *
+   * @param receipt the transaction receipt to build RLP from.
+   * @return the RLP.
+   */
+  Bytes buildRlpFromReceipt(final TransactionReceipt receipt);
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/sync/SynchronizationService.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.sync;
+
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.plugin.data.BlockBody;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.services.BesuService;
+
+/** Synchronization service wraps the sync state and sync event lifecycle. */
+public interface SynchronizationService extends BesuService {
+
+  /**
+   * Enables P2P discovery.
+   *
+   * @param head the head of the chain.
+   * @param safeBlock the safe block.
+   * @param finalizedBlock the finalized block.
+   */
+  void fireNewUnverifiedForkchoiceEvent(Hash head, Hash safeBlock, Hash finalizedBlock);
+
+  /**
+   * Set the head of the chain.
+   *
+   * @param blockHeader the block header
+   * @param blockBody the block body
+   * @return true if the head was set, false otherwise.
+   */
+  boolean setHead(final BlockHeader blockHeader, final BlockBody blockBody);
+
+  /**
+   * Adds the block header and body to the head of the chain directly, without using a block
+   * importer or validation.
+   *
+   * @param blockHeader the block header
+   * @param blockBody the block body
+   * @return true if the head was set, false otherwise.
+   */
+  boolean setHeadUnsafe(BlockHeader blockHeader, BlockBody blockBody);
+
+  /**
+   * Returns whether the initial chain and worldstate sync is complete.
+   *
+   * @return true if the initial sync phase is done, false otherwise.
+   */
+  boolean isInitialSyncPhaseDone();
+
+  /** Disables the worldstate trie for update. */
+  void disableWorldStateTrie();
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/sync/WorldStateConfiguration.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.sync;
+
+/** interface for worldstate configuration * */
+public interface WorldStateConfiguration {
+
+  /**
+   * Returns whether the trie is disabled.
+   *
+   * @return true if the trie is disabled, false otherwise.
+   */
+  boolean isTrieDisabled();
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/transactionpool/TransactionPoolService.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.transactionpool;
+
+import org.hyperledger.besu.plugin.services.BesuService;
+
+/** Service to enable and disable the transaction pool. */
+public interface TransactionPoolService extends BesuService {
+  /** Enables the transaction pool. */
+  void disableTransactionPool();
+
+  /** Disables the transaction pool. */
+  void enableTransactionPool();
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/trielogs/TrieLogProvider.java

@@ -19,8 +19,19 @@ import org.hyperledger.besu.datatypes.Hash;
 import java.util.List;
 import java.util.Optional;
 
+import org.apache.tuweni.bytes.Bytes;
+
 /** Trielog provider interface for a given block hash. */
 public interface TrieLogProvider {
+  /**
+   * Saves the TrieLog layer for the given block hash.
+   *
+   * @param blockHash the block hash
+   * @param blockNumber the block number
+   * @param trieLog the associated TrieLog layer
+   */
+  void saveRawTrieLogLayer(final Hash blockHash, final long blockNumber, final Bytes trieLog);
+
   /**
    * Returns the TrieLog layer for the given block hash.
    *
@@ -30,6 +41,14 @@ public interface TrieLogProvider {
    */
   <T extends TrieLog.LogTuple<?>> Optional<TrieLog> getTrieLogLayer(final Hash blockHash);
 
+  /**
+   * Get the raw TrieLog layer for the given block hash.
+   *
+   * @param blockHash the block hash
+   * @return the raw TrieLog layer bytes for the given block hash
+   */
+  Optional<Bytes> getRawTrieLogLayer(final Hash blockHash);
+
   /**
    * Returns the TrieLog layer for the given block number.
    *
@@ -39,6 +58,14 @@ public interface TrieLogProvider {
    */
   <T extends TrieLog.LogTuple<?>> Optional<TrieLog> getTrieLogLayer(final long blockNumber);
 
+  /**
+   * Get the raw TrieLog layer for the given block number.
+   *
+   * @param blockNumber the block number
+   * @return the raw TrieLog layer bytes for the given block number
+   */
+  Optional<Bytes> getRawTrieLogLayer(final long blockNumber);
+
   /**
    * Returns the TrieLog layers for the given block number range.
    *