@Test void shouldRetryWhenEth1ChainIsLessThanEth1FollowDistance() { final BigInteger firstEth1BlockNumber = BigInteger.valueOf(60); final Block firstLatestBlock = block(firstEth1BlockNumber, 100000); final BigInteger secondEth1BlockNumber = BigInteger.valueOf(101); final Block secondLatestBlock = block(secondEth1BlockNumber, 150000); when(eth1Provider.getLatestEth1Block()) .thenReturn(SafeFuture.completedFuture(firstLatestBlock)) .thenReturn(SafeFuture.completedFuture(secondLatestBlock)); Constants.ETH1_FOLLOW_DISTANCE = UInt64.valueOf(100); when(eth1DepositStorageChannel.replayDepositEvents()).thenReturn(NOTHING_REPLAYED); manager.start(); inOrder.verify(eth1DepositStorageChannel).replayDepositEvents(); verify(eth1Provider).getLatestEth1Block(); verify(eth1Provider, never()).getGuaranteedEth1Block((UInt64) any()); assertThat(asyncRunner.hasDelayedActions()).isTrue(); asyncRunner.executeQueuedActions(); verify(eth1Provider, times(2)).getLatestEth1Block(); verify(eth1Provider).getGuaranteedEth1Block((UInt64) any()); assertNoUncaughtExceptions(); }
public void start() { eth1DepositStorageChannel .replayDepositEvents() .thenCompose( replayDepositsResult -> { replayDepositsResult .getLastProcessedDepositIndex() .map(UInt64::valueOf) .ifPresent(eth1EventsPublisher::setLatestPublishedDeposit); return getHead() .thenCompose(headBlock -> processStart(headBlock, replayDepositsResult)); }) .thenAccept(__ -> LOG.info("Eth1DepositsManager successfully ran startup sequence.")) .exceptionally( (err) -> { throw new FatalServiceFailureException(getClass(), err); }) .reportExceptions(); }
Eth1DepositManager { public void start() { eth1DepositStorageChannel .replayDepositEvents() .thenCompose( replayDepositsResult -> { replayDepositsResult .getLastProcessedDepositIndex() .map(UInt64::valueOf) .ifPresent(eth1EventsPublisher::setLatestPublishedDeposit); return getHead() .thenCompose(headBlock -> processStart(headBlock, replayDepositsResult)); }) .thenAccept(__ -> LOG.info("Eth1DepositsManager successfully ran startup sequence.")) .exceptionally( (err) -> { throw new FatalServiceFailureException(getClass(), err); }) .reportExceptions(); } }
Eth1DepositManager { public void start() { eth1DepositStorageChannel .replayDepositEvents() .thenCompose( replayDepositsResult -> { replayDepositsResult .getLastProcessedDepositIndex() .map(UInt64::valueOf) .ifPresent(eth1EventsPublisher::setLatestPublishedDeposit); return getHead() .thenCompose(headBlock -> processStart(headBlock, replayDepositsResult)); }) .thenAccept(__ -> LOG.info("Eth1DepositsManager successfully ran startup sequence.")) .exceptionally( (err) -> { throw new FatalServiceFailureException(getClass(), err); }) .reportExceptions(); } Eth1DepositManager(
      final Eth1Provider eth1Provider,
      final AsyncRunner asyncRunner,
      final ValidatingEth1EventsPublisher eth1EventsPublisher,
      final Eth1DepositStorageChannel eth1DepositStorageChannel,
      final DepositProcessingController depositProcessingController,
      final MinimumGenesisTimeBlockFinder minimumGenesisTimeBlockFinder); }
Eth1DepositManager { public void start() { eth1DepositStorageChannel .replayDepositEvents() .thenCompose( replayDepositsResult -> { replayDepositsResult .getLastProcessedDepositIndex() .map(UInt64::valueOf) .ifPresent(eth1EventsPublisher::setLatestPublishedDeposit); return getHead() .thenCompose(headBlock -> processStart(headBlock, replayDepositsResult)); }) .thenAccept(__ -> LOG.info("Eth1DepositsManager successfully ran startup sequence.")) .exceptionally( (err) -> { throw new FatalServiceFailureException(getClass(), err); }) .reportExceptions(); } Eth1DepositManager(
      final Eth1Provider eth1Provider,
      final AsyncRunner asyncRunner,
      final ValidatingEth1EventsPublisher eth1EventsPublisher,
      final Eth1DepositStorageChannel eth1DepositStorageChannel,
      final DepositProcessingController depositProcessingController,
      final MinimumGenesisTimeBlockFinder minimumGenesisTimeBlockFinder); void start(); void stop(); }
Eth1DepositManager { public void start() { eth1DepositStorageChannel .replayDepositEvents() .thenCompose( replayDepositsResult -> { replayDepositsResult .getLastProcessedDepositIndex() .map(UInt64::valueOf) .ifPresent(eth1EventsPublisher::setLatestPublishedDeposit); return getHead() .thenCompose(headBlock -> processStart(headBlock, replayDepositsResult)); }) .thenAccept(__ -> LOG.info("Eth1DepositsManager successfully ran startup sequence.")) .exceptionally( (err) -> { throw new FatalServiceFailureException(getClass(), err); }) .reportExceptions(); } Eth1DepositManager(
      final Eth1Provider eth1Provider,
      final AsyncRunner asyncRunner,
      final ValidatingEth1EventsPublisher eth1EventsPublisher,
      final Eth1DepositStorageChannel eth1DepositStorageChannel,
      final DepositProcessingController depositProcessingController,
      final MinimumGenesisTimeBlockFinder minimumGenesisTimeBlockFinder); void start(); void stop();  }