private void downloadAvailableWorldStateFromPeers(
      final int peerCount,
      final int accountCount,
      final int hashesPerRequest,
      final int maxOutstandingRequests,
      final NetworkResponder networkResponder) {
    final int trailingPeerCount = 5;

    // Setup "remote" state
    final ForestWorldStateKeyValueStorage remoteStorage =
        new ForestWorldStateKeyValueStorage(new InMemoryKeyValueStorage());
    final WorldStateArchive remoteWorldStateArchive =
        new ForestWorldStateArchive(
            new WorldStateStorageCoordinator(remoteStorage),
            createPreimageStorage(),
            EvmConfiguration.DEFAULT);
    final MutableWorldState remoteWorldState = remoteWorldStateArchive.getWorldState();

    // Generate accounts and save corresponding state root
    final List<Account> accounts = dataGen.createRandomAccounts(remoteWorldState, accountCount);
    final Hash stateRoot = remoteWorldState.rootHash();
    assertThat(stateRoot).isNotEqualTo(EMPTY_TRIE_ROOT); // Sanity check
    final BlockHeader header =
        dataGen.block(BlockOptions.create().setStateRoot(stateRoot).setBlockNumber(10)).getHeader();

    // Generate more data that should not be downloaded
    final List<Account> otherAccounts = dataGen.createRandomAccounts(remoteWorldState, 5);
    final Hash otherStateRoot = remoteWorldState.rootHash();
    final BlockHeader otherHeader =
        dataGen
            .block(BlockOptions.create().setStateRoot(otherStateRoot).setBlockNumber(11))
            .getHeader();
    assertThat(otherStateRoot).isNotEqualTo(stateRoot); // Sanity check

    final InMemoryTasksPriorityQueues<NodeDataRequest> taskCollection =
        new InMemoryTasksPriorityQueues<>();
    final ForestWorldStateKeyValueStorage localStorage =
        new ForestWorldStateKeyValueStorage(new InMemoryKeyValueStorage());
    final WorldStateArchive localWorldStateArchive =
        new ForestWorldStateArchive(
            new WorldStateStorageCoordinator(localStorage),
            createPreimageStorage(),
            EvmConfiguration.DEFAULT);
    final SynchronizerConfiguration syncConfig =
        SynchronizerConfiguration.builder()
            .worldStateHashCountPerRequest(hashesPerRequest)
            .worldStateRequestParallelism(maxOutstandingRequests)
            .build();
    final WorldStateDownloader downloader =
        createDownloader(syncConfig, ethProtocolManager.ethContext(), localStorage, taskCollection);

    // Create some peers that can respond
    final List<RespondingEthPeer> usefulPeers =
        Stream.generate(
                () -> EthProtocolManagerTestUtil.createPeer(ethProtocolManager, header.getNumber()))
            .limit(peerCount)
            .collect(Collectors.toList());
    // And some irrelevant peers
    final List<RespondingEthPeer> trailingPeers =
        Stream.generate(
                () ->
                    EthProtocolManagerTestUtil.createPeer(
                        ethProtocolManager, header.getNumber() - 1L))
            .limit(trailingPeerCount)
            .toList();

    // Start downloader
    final CompletableFuture<?> result = downloader.run(null, new FastSyncState(header));
    // A second run should return an error without impacting the first result
    final CompletableFuture<?> secondResult = downloader.run(null, new FastSyncState(header));
    assertThat(secondResult).isCompletedExceptionally();
    assertThat(result).isNotCompletedExceptionally();

    // Respond to node data requests
    // Send one round of full responses, so that we can get multiple requests queued up
    final RespondingEthPeer.Responder fullResponder =
        RespondingEthPeer.blockchainResponder(mock(Blockchain.class), remoteWorldStateArchive);
    for (final RespondingEthPeer peer : usefulPeers) {
      peer.respond(fullResponder);
    }
    // Respond to remaining queued requests in custom way
    networkResponder.respond(usefulPeers, remoteWorldStateArchive, result);

    // Check that trailing peers were not queried for data
    for (final RespondingEthPeer trailingPeer : trailingPeers) {
      assertThat(trailingPeer.hasOutstandingRequests()).isFalse();
    }

    // Check that all expected account data was downloaded
    final WorldState localWorldState = localWorldStateArchive.get(stateRoot, null).get();
    assertThat(result).isDone();
    assertAccountsMatch(localWorldState, accounts);

    // We shouldn't have any extra data locally
    assertThat(localStorage.contains(otherHeader.getStateRoot())).isFalse();
    for (final Account otherAccount : otherAccounts) {
      assertThat(localWorldState.get(otherAccount.getAddress())).isNull();
    }
  }
Returns the first set of node hashes that would need to be requested from the network after
retrieving the root node in order to rebuild the trie represented by the given rootHash and
storage.

@param storage Storage holding node data required to reconstitute the trie represented by
    rootHash
@param rootHash The hash of the root node of some trie
@return A list of node hashes
ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/FastWorldStateDownloaderTest.java