@Test void sync_failSyncIfPeerThrottlesTooAggressively() { final UInt64 startSlot = UInt64.ONE; UInt64 minPeerSlot = MAX_BLOCK_BY_RANGE_REQUEST_SIZE.plus(startSlot); withPeerFinalizedEpoch(compute_epoch_at_slot(minPeerSlot)); final List<SafeFuture<Void>> requestFutures = new ArrayList<>(); OngoingStubbing<SafeFuture<Void>> requestStub = when(peer.requestBlocksByRange(any(), any(), any(), any())); for (int i = 0; i < PeerSync.MAX_THROTTLED_REQUESTS + 1; i++) { final SafeFuture<Void> future = new SafeFuture<>(); requestStub = requestStub.thenReturn(future); requestFutures.add(future); } final SafeFuture<PeerSyncResult> syncFuture = peerSync.sync(peer); assertThat(syncFuture).isNotDone(); verify(peer) .requestBlocksByRange( eq(startSlot), eq(MAX_BLOCK_BY_RANGE_REQUEST_SIZE), eq(UInt64.ONE), responseListenerArgumentCaptor.capture()); int nextBlock = startSlot.intValue(); for (int i = 0; i < PeerSync.MAX_THROTTLED_REQUESTS; i++) { completeRequestWithBlockAtSlot(requestFutures.get(i), nextBlock); nextBlock += 1; } assertThat(syncFuture).isNotDone(); final int lastRequestIndex = PeerSync.MAX_THROTTLED_REQUESTS; completeRequestWithBlockAtSlot(requestFutures.get(lastRequestIndex), nextBlock + 2); assertThat(syncFuture).isCompletedWithValue(PeerSyncResult.EXCESSIVE_THROTTLING); verify(peer, never()).disconnectCleanly(any()); }
public SafeFuture<PeerSyncResult> sync(final Eth2Peer peer) { LOG.debug("Start syncing to peer {}", peer); final UInt64 finalizedEpoch = storageClient.getFinalizedEpoch(); final UInt64 latestFinalizedSlot = compute_start_slot_at_epoch(finalizedEpoch); final UInt64 firstNonFinalSlot = latestFinalizedSlot.plus(UInt64.ONE); this.startingSlot = firstNonFinalSlot; return executeSync(peer, firstNonFinalSlot, SafeFuture.COMPLETE, true) .whenComplete( (res, err) -> { if (err != null) { LOG.debug("Failed to sync with peer {}: {}", peer, err); } else { LOG.debug("Finished syncing (with status {}) to peer {}", res.name(), peer); } }); }
PeerSync { public SafeFuture<PeerSyncResult> sync(final Eth2Peer peer) { LOG.debug("Start syncing to peer {}", peer); final UInt64 finalizedEpoch = storageClient.getFinalizedEpoch(); final UInt64 latestFinalizedSlot = compute_start_slot_at_epoch(finalizedEpoch); final UInt64 firstNonFinalSlot = latestFinalizedSlot.plus(UInt64.ONE); this.startingSlot = firstNonFinalSlot; return executeSync(peer, firstNonFinalSlot, SafeFuture.COMPLETE, true) .whenComplete( (res, err) -> { if (err != null) { LOG.debug("Failed to sync with peer {}: {}", peer, err); } else { LOG.debug("Finished syncing (with status {}) to peer {}", res.name(), peer); } }); } }
PeerSync { public SafeFuture<PeerSyncResult> sync(final Eth2Peer peer) { LOG.debug("Start syncing to peer {}", peer); final UInt64 finalizedEpoch = storageClient.getFinalizedEpoch(); final UInt64 latestFinalizedSlot = compute_start_slot_at_epoch(finalizedEpoch); final UInt64 firstNonFinalSlot = latestFinalizedSlot.plus(UInt64.ONE); this.startingSlot = firstNonFinalSlot; return executeSync(peer, firstNonFinalSlot, SafeFuture.COMPLETE, true) .whenComplete( (res, err) -> { if (err != null) { LOG.debug("Failed to sync with peer {}: {}", peer, err); } else { LOG.debug("Finished syncing (with status {}) to peer {}", res.name(), peer); } }); } PeerSync(
      final AsyncRunner asyncRunner,
      final RecentChainData storageClient,
      final BlockImporter blockImporter,
      final MetricsSystem metricsSystem); }
PeerSync { public SafeFuture<PeerSyncResult> sync(final Eth2Peer peer) { LOG.debug("Start syncing to peer {}", peer); final UInt64 finalizedEpoch = storageClient.getFinalizedEpoch(); final UInt64 latestFinalizedSlot = compute_start_slot_at_epoch(finalizedEpoch); final UInt64 firstNonFinalSlot = latestFinalizedSlot.plus(UInt64.ONE); this.startingSlot = firstNonFinalSlot; return executeSync(peer, firstNonFinalSlot, SafeFuture.COMPLETE, true) .whenComplete( (res, err) -> { if (err != null) { LOG.debug("Failed to sync with peer {}: {}", peer, err); } else { LOG.debug("Finished syncing (with status {}) to peer {}", res.name(), peer); } }); } PeerSync(
      final AsyncRunner asyncRunner,
      final RecentChainData storageClient,
      final BlockImporter blockImporter,
      final MetricsSystem metricsSystem); SafeFuture<PeerSyncResult> sync(final Eth2Peer peer); void stop(); UInt64 getStartingSlot(); }
PeerSync { public SafeFuture<PeerSyncResult> sync(final Eth2Peer peer) { LOG.debug("Start syncing to peer {}", peer); final UInt64 finalizedEpoch = storageClient.getFinalizedEpoch(); final UInt64 latestFinalizedSlot = compute_start_slot_at_epoch(finalizedEpoch); final UInt64 firstNonFinalSlot = latestFinalizedSlot.plus(UInt64.ONE); this.startingSlot = firstNonFinalSlot; return executeSync(peer, firstNonFinalSlot, SafeFuture.COMPLETE, true) .whenComplete( (res, err) -> { if (err != null) { LOG.debug("Failed to sync with peer {}: {}", peer, err); } else { LOG.debug("Finished syncing (with status {}) to peer {}", res.name(), peer); } }); } PeerSync(
      final AsyncRunner asyncRunner,
      final RecentChainData storageClient,
      final BlockImporter blockImporter,
      final MetricsSystem metricsSystem); SafeFuture<PeerSyncResult> sync(final Eth2Peer peer); void stop(); UInt64 getStartingSlot();  }