@Test void shouldRetryIfEth1NodeIsNotReady() { final int retryCount = 10; final BigInteger headBlockNumber = BigInteger.valueOf(100); when(eth1DepositStorageChannel.replayDepositEvents()).thenReturn(NOTHING_REPLAYED); when(eth1Provider.getLatestEth1Block()) .thenReturn(SafeFuture.failedFuture(new IllegalStateException("Connection refused"))); when(depositProcessingController.fetchDepositsInRange(any(), any())).thenReturn(COMPLETE); manager.start(); for (int i = 0; i < retryCount; i++) { assertThat(asyncRunner.countDelayedActions()).isEqualTo(1); asyncRunner.executeQueuedActions(); } withFollowDistanceHead(headBlockNumber, MIN_GENESIS_BLOCK_TIMESTAMP - 1); assertThat(asyncRunner.countDelayedActions()).isEqualTo(1); asyncRunner.executeQueuedActions(); inOrder.verify(eth1DepositStorageChannel).replayDepositEvents(); inOrder .verify(depositProcessingController) .fetchDepositsInRange(BigInteger.ZERO, headBlockNumber); inOrder.verify(depositProcessingController).switchToBlockByBlockMode(); inOrder.verify(depositProcessingController).startSubscription(BigInteger.valueOf(101)); inOrder.verifyNoMoreInteractions(); 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();  }