diff --git a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/SyncConfig.java b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/SyncConfig.java index 7f78013f1c2..119f13b0772 100644 --- a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/SyncConfig.java +++ b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/SyncConfig.java @@ -29,8 +29,8 @@ public class SyncConfig { public static final int DEFAULT_FORWARD_SYNC_MAX_PENDING_BATCHES = 5; /** - * Must be >= FORWARD_SYNC_BATCH_SIZE * FORWARD_SYNC_MAX_PENDING_BATCHES to avoid evicting - * completed trackers before the sync pipeline imports them. + * Sized for the default linear forward-sync pipeline. Trackers still needed for import may exceed + * this soft limit, up to the sampler's hard limit. */ public static final int DEFAULT_MAX_RECENTLY_SAMPLED_BLOCKS = 128; diff --git a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSync.java b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSync.java index db71ce0654e..6cdb74d0e35 100644 --- a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSync.java +++ b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSync.java @@ -391,7 +391,16 @@ private void startNextImport() { final SafeFuture importFuture = batchImporter.importBatch(batch); importingBatchFuture = Optional.of(importFuture); importFuture - .thenAcceptAsync(result -> onImportComplete(result, batch), eventThread) + .handleAsync( + (result, error) -> { + if (error != null) { + onImportFailed(batch, error); + } else { + onImportComplete(result, batch); + } + return null; + }, + eventThread) .propagateExceptionTo(syncResult); }); } @@ -416,6 +425,23 @@ private void markBatchesAsContested(final NavigableSet contestedBatches) contestedBatches.forEach(Batch::markAsContested); } + /** + * Handles an import which completed exceptionally, for example because it failed unexpectedly. + * The batch didn't import, so it is handled as any other failed import - importing state has to + * be released either way, otherwise the sync would wait forever for an import that already + * finished. + */ + private void onImportFailed(final Batch importedBatch, final Throwable error) { + eventThread.checkOnEventThread(); + if (!isCurrentlyImportingBatch(importedBatch)) { + // already released, e.g. the sync was aborted which cancelled the import + return; + } + // an import should never fail exceptionally, so make it visible rather than only retrying + LOG.warn("Import of batch {} failed unexpectedly", importedBatch, error); + onImportComplete(BatchImportResult.IMPORT_FAILED, importedBatch); + } + private void onImportComplete( final BatchImporter.BatchImportResult result, final Batch importedBatch) { eventThread.checkOnEventThread(); @@ -448,7 +474,11 @@ private void onImportComplete( } else if (result == BatchImportResult.EXECUTION_CLIENT_OFFLINE || result == BatchImportResult.DATA_NOT_AVAILABLE) { if (!scheduledProgressSync) { - LOG.warn("Unable to import blocks: {}", result); + LOG.warn( + "Unable to import blocks ({} - {}): {}", + importedBatch.getFirstSlot(), + importedBatch.getLastSlot(), + result); asyncRunner .runAfterDelay( () -> diff --git a/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSyncTest.java b/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSyncTest.java index e7019088c6f..499e37e82e4 100644 --- a/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSyncTest.java +++ b/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSyncTest.java @@ -18,6 +18,7 @@ import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -170,6 +171,31 @@ void shouldCancelInProgressImportWhenAborted() { assertThat(importResult).isCancelled(); } + @Test + void shouldHandleImportCompletingExceptionallyAsFailedImport() { + final SignedBlockAndState block5 = chainBuilder.generateBlockAtSlot(5); + final SignedBlockAndState block26 = chainBuilder.generateBlockAtSlot(26); + final SafeFuture syncResult = sync.syncToChain(targetChain); + + final Batch batch0 = batches.get(0); + final Batch batch1 = batches.get(1); + batches.receiveBlocks(batch0, block5.getBlock()); + batches.receiveBlocks(batch1, block26.getBlock()); + final SafeFuture importResult = batches.getImportResult(batch0); + + importResult.completeExceptionally(new RuntimeException("Import blew up")); + + // Treated as a failed import: batches forming a chain with it are invalid and get re-requested + batches.assertMarkedInvalid(batch0); + batches.assertMarkedInvalid(batch1); + assertThat(syncResult).isNotDone(); + + // And the importing state was released so the re-requested batch can be imported again + batches.receiveBlocks(batch0, block5.getBlock()); + batches.receiveBlocks(batch1, block26.getBlock()); + verify(batchImporter, times(2)).importBatch(batches.getEventThreadOnlyBatch(batch0)); + } + @Test void shouldNotImportPreviousBatchWhenBoundaryParentExecutionPayloadIsMissing() { final UInt64 gloasForkEpoch = UInt64.valueOf(1000); @@ -944,6 +970,44 @@ void shouldDelaySwitchingToNewChainUntilCurrentImportCompletes() { assertThat(batches.get(0).getFirstSlot()).isEqualTo(finalizedBlock.getSlot()); } + @Test + void shouldSwitchToNewChainWhenDelayedImportCompletesExceptionally() { + assertThat(sync.syncToChain(targetChain)).isNotDone(); + + final Batch batch0 = batches.get(0); + final Batch batch1 = batches.get(1); + batches.receiveBlocks(batch0, chainBuilder.generateBlockAtSlot(1).getBlock()); + batches.receiveBlocks( + batch1, chainBuilder.generateBlockAtSlot(batch1.getFirstSlot()).getBlock()); + + assertBatchImported(batch0); + + final Batch batch4 = batches.get(4); + + // Switch to a new chain while batch0 is still importing + targetChain = chainWith(dataStructureUtil.randomSlotAndBlockRoot(), syncSource); + final SafeFuture newSyncResult = sync.syncToChain(targetChain); + assertThat(newSyncResult).isNotDone(); + + // And return blocks so the new chain doesn't match up. + batches.receiveBlocks( + batch4, chainBuilder.generateBlockAtSlot(batch4.getLastSlot()).getBlock()); + final Batch batch5 = batches.get(5); + batches.receiveBlocks(batch5, dataStructureUtil.randomSignedBeaconBlock(batch5.getFirstSlot())); + + // All batches should have been dropped and none started until the import completes + batches.forEach(this::assertBatchNotActive); + batches.clearBatchList(); + + final SignedBlockAndState finalizedBlock = storageSystem.chainUpdater().finalizeEpoch(1); + batches.getImportResult(batch0).completeExceptionally(new RuntimeException("Import blew up")); + + // The failure belongs to the chain we already left, so the new sync continues downloading + // from the latest finalized checkpoint instead of failing or stalling + assertThat(newSyncResult).isNotDone(); + assertThat(batches.get(0).getFirstSlot()).isEqualTo(finalizedBlock.getSlot()); + } + @Test void shouldProgressWhenThereAreManyEmptyBatchesInARow() { assertThat(sync.syncToChain(targetChain)).isNotDone(); diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicImpl.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicImpl.java index 5b3bfcdd816..ba80a4e0173 100644 --- a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicImpl.java +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicImpl.java @@ -262,16 +262,17 @@ private void makeRoomForNewTracker() { if (currentSize < maxRecentlySampledBlocks) { return; } - // First pass: evict completed trackers, oldest-created first. Best-effort: CHM size/iteration - // are weakly consistent and concurrent writers may shift counts while we work. + // First pass: evict trackers no longer needed for import, oldest-created first. Best-effort: + // CHM size/iteration are weakly consistent and concurrent writers may shift counts while we + // work. final int softExcess = currentSize - maxRecentlySampledBlocks + 1; recentlySampledColumnsByRoot.entrySet().stream() - .filter(e -> e.getValue().completionFuture().isDone()) + .filter(e -> isSafeToEvict(e.getValue())) .sorted(Comparator.comparingLong(e -> e.getValue().createdAtNanos())) .limit(softExcess) .forEach(e -> recentlySampledColumnsByRoot.remove(e.getKey(), e.getValue())); - // Hard cap: if we're still at 4x the limit even after evicting completed trackers, - // force-evict the oldest incomplete ones to prevent unbounded growth. + // Hard cap: if we're still at 4x the limit after safe eviction, force-evict the oldest + // remaining trackers to prevent unbounded growth. final int hardLimit = maxRecentlySampledBlocks * 4; final int afterSoft = recentlySampledColumnsByRoot.size(); if (afterSoft < hardLimit) { @@ -299,6 +300,12 @@ private void makeRoomForNewTracker() { }); } + private boolean isSafeToEvict(final DataColumnSamplingTracker tracker) { + return tracker.completionFuture().isCompletedExceptionally() + || (tracker.completionFuture().isDone() + && isDataAvailabilityAlreadySatisfied(tracker.slot(), tracker.blockRoot())); + } + private SafeFuture retrieveColumnWithSamplingAndCustody( final DataColumnSlotAndIdentifier id, final DataColumnSamplingTracker tracker) { return retriever diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetriever.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetriever.java index de0582e9cc0..e500f2f0691 100644 --- a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetriever.java +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetriever.java @@ -21,6 +21,7 @@ import java.util.Optional; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -140,6 +141,10 @@ private Stream matchRequestsAndPeers() { } private boolean activateMatchedRequest(final RequestMatch match) { + // The request may have completed after it was matched to a peer. + if (isStaleRequest(match.request)) { + return false; + } if (!match.request.activeRpcRequestSet.compareAndSet(false, true)) { // already activated return false; @@ -155,6 +160,13 @@ private boolean activateMatchedRequest(final RequestMatch match) { reqRespCompleted(match.request, sidecar); if (err == null) { match.peer.countSidecarReceived(); + } else if (ExceptionUtil.hasCause(err, CancellationException.class)) { + // the request was cancelled by us because the sidecar was no longer needed (for + // example it arrived via gossip), so the peer must not be penalised for it + match.peer.discardSidecarRequest(); + LOG.trace( + "SimpleSidecarRetriever.Request cancelled for {}", + () -> match.request.columnId); } else { LOG.debug( "SimpleSidecarRetriever.Request failed for {} due to: {}", @@ -169,10 +181,20 @@ private boolean activateMatchedRequest(final RequestMatch match) { // log all the info to fix the bug activeRpcRequest.ignoreCancelException().finishStackTrace(); - match.request.activeRpcRequest = new ActiveRequest(activeRpcRequest, match.peer); + match.request.activeRpcRequest = new ActiveRequest(reqRespPromise, match.peer); + // The request may complete while its RPC is being set up. + if (isStaleRequest(match.request)) { + reqRespPromise.cancel(true); + return false; + } return true; } + @SuppressWarnings({"ReferenceEquality", "ReferenceComparison"}) + private boolean isStaleRequest(final RetrieveRequest request) { + return pendingRequests.get(request.columnId) != request; + } + private Optional findBestMatchingPeer( final RetrieveRequest request, final RequestTracker ongoingRequestsTracker) { final Stream matchingPeers = findMatchingPeers(request, ongoingRequestsTracker); @@ -237,8 +259,14 @@ private void nextRound() { private void reqRespCompleted( final RetrieveRequest request, final DataColumnSidecar maybeResult) { - if (maybeResult != null && pendingRequests.remove(request.columnId) != null) { + if (maybeResult != null && pendingRequests.remove(request.columnId, request)) { + final ActiveRequest activeRequest = request.activeRpcRequest; + request.activeRpcRequest = null; + request.activeRpcRequestSet.set(false); request.result.completeAsync(maybeResult, asyncRunner); + if (activeRequest != null) { + activeRequest.promise().cancel(true); + } retrieveCounter.incrementAndGet(); } else if (request.activeRpcRequestSet.compareAndSet(true, false)) { request.activeRpcRequest = null; @@ -300,7 +328,7 @@ Map getConnectedPeers() { return connectedPeers; } - private record ActiveRequest(SafeFuture promise, ConnectedPeer peer) {} + private record ActiveRequest(SafeFuture promise, ConnectedPeer peer) {} private static class RetrieveRequest { final DataColumnSlotAndIdentifier columnId; @@ -373,6 +401,15 @@ public void countSidecarRequest() { } } + /** + * Reverts a {@link #countSidecarRequest()} for a request which was cancelled on our side, so + * that the peer's response score is not affected by a response we stopped waiting for. + */ + public void discardSidecarRequest() { + // counters could have been reset in between, so never go below the initial value + sidecarsRequested.updateAndGet(current -> Math.max(1, current - 1)); + } + private void resetCounters() { sidecarsRequested.set(1); sidecarsReceived.set(1); diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityChecker.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityChecker.java index 4ed991e9e40..d33653647d7 100644 --- a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityChecker.java +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityChecker.java @@ -13,7 +13,6 @@ package tech.pegasys.teku.statetransition.forkchoice; -import com.google.common.annotations.VisibleForTesting; import java.time.Duration; import java.util.List; import java.util.Optional; @@ -41,7 +40,7 @@ public class DataColumnSidecarAvailabilityChecker implements AvailabilityChecker private final RecentChainData recentChainData; private final SignedBeaconBlock block; private final Optional signedEnvelope; - private final Duration waitForSamplerCompletionTimeout; + private static final long BATCH_SYNC_TIMEOUT_BOOST = 5L; public DataColumnSidecarAvailabilityChecker( final DataAvailabilitySampler dataAvailabilitySampler, @@ -71,22 +70,6 @@ private DataColumnSidecarAvailabilityChecker( this.recentChainData = recentChainData; this.block = block; this.signedEnvelope = signedEnvelope; - this.waitForSamplerCompletionTimeout = calculateCompletionTimeout(spec, block.getSlot()); - } - - @VisibleForTesting - DataColumnSidecarAvailabilityChecker( - final DataAvailabilitySampler dataAvailabilitySampler, - final Spec spec, - final RecentChainData recentChainData, - final SignedBeaconBlock block, - final Duration waitForSamplerCompletionTimeout) { - this.dataAvailabilitySampler = dataAvailabilitySampler; - this.spec = spec; - this.recentChainData = recentChainData; - this.block = block; - this.signedEnvelope = Optional.empty(); - this.waitForSamplerCompletionTimeout = waitForSamplerCompletionTimeout; } @Override @@ -115,7 +98,7 @@ public boolean initiateDataAvailabilityCheck() { .checkDataAvailability(block.getSlot(), block.getRoot()) .propagateTo(localFuture); localFuture - .orTimeout(waitForSamplerCompletionTimeout) + .orTimeout(calculateCompletionTimeout(spec, block.getSlot())) .thenApply(DataAndValidationResult::validResult) .exceptionallyCompose( error -> @@ -174,7 +157,16 @@ private boolean isBlockOutsideDataAvailabilityWindow() { recentChainData.getStore(), block.getSlot()); } - static Duration calculateCompletionTimeout(final Spec spec, final UInt64 slot) { + private Duration calculateCompletionTimeout(final Spec spec, final UInt64 slot) { + if (recentChainData.getCurrentSlot().isEmpty()) { + return Duration.ofMillis(spec.getSlotDurationMillis(slot)); + } + final UInt64 currentSlot = recentChainData.getCurrentSlot().get(); + final UInt64 recentEpochSlot = currentSlot.minusMinZero(spec.getSlotsPerEpoch(currentSlot)); + // BatchSync downloads 125 slots simultaneously. We should increase timeout accordingly + if (slot.isLessThan(recentEpochSlot)) { + return Duration.ofMillis(spec.getSlotDurationMillis(slot) * BATCH_SYNC_TIMEOUT_BOOST); + } return Duration.ofMillis(spec.getSlotDurationMillis(slot)); } } diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicTest.java index e00cb247e68..a827ee1f150 100644 --- a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicTest.java +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/DasSamplerBasicTest.java @@ -691,6 +691,8 @@ void getOrCreateTracker_shouldEvictOldestCompletedTrackerAtSoftLimit() { .put(oldestCompleted.blockRoot(), oldestCompleted); smallSampler.getRecentlySampledColumnsByRoot().put(newerCompleted.blockRoot(), newerCompleted); smallSampler.getRecentlySampledColumnsByRoot().put(incomplete.blockRoot(), incomplete); + when(recentChainData.containsBlock(oldestCompleted.blockRoot())).thenReturn(true); + when(recentChainData.containsBlock(newerCompleted.blockRoot())).thenReturn(true); // Trigger insertion of a new tracker, which calls makeRoomForNewTracker final DataColumnSidecar sidecar = @@ -721,6 +723,8 @@ void getOrCreateTracker_shouldPreferOldestCompletedWhenSoftEvicting() { .getRecentlySampledColumnsByRoot() .put(oldestCompleted.blockRoot(), oldestCompleted); smallSampler.getRecentlySampledColumnsByRoot().put(newerCompleted.blockRoot(), newerCompleted); + when(recentChainData.containsBlock(oldestCompleted.blockRoot())).thenReturn(true); + when(recentChainData.containsBlock(newerCompleted.blockRoot())).thenReturn(true); final DataColumnSidecar sidecar = dataStructureUtil.randomDataColumnSidecar( @@ -736,6 +740,27 @@ void getOrCreateTracker_shouldPreferOldestCompletedWhenSoftEvicting() { .containsKey(sidecar.getBeaconBlockRoot()); } + @Test + void getOrCreateTracker_shouldRetainCompletedTrackerAwaitingImportAtSoftLimit() { + final DasSamplerBasicImpl smallSampler = createSampler(2, new StubMetricsSystem()); + final DataColumnSamplingTracker completedPendingImport = + completedMockTracker(dataStructureUtil.randomBytes32(), 100L); + smallSampler + .getRecentlySampledColumnsByRoot() + .put(completedPendingImport.blockRoot(), completedPendingImport); + + final DataColumnSidecar sidecar = + dataStructureUtil.randomDataColumnSidecar( + dataStructureUtil.randomSignedBeaconBlockHeader(), SAMPLING_INDEX_0); + when(rpcFetchDelayProvider.calculate(sidecar.getSlot())).thenReturn(Duration.ofSeconds(1)); + + smallSampler.onNewValidatedDataColumnSidecar(sidecar, RemoteOrigin.GOSSIP); + + assertThat(smallSampler.getRecentlySampledColumnsByRoot()) + .containsKey(completedPendingImport.blockRoot()) + .containsKey(sidecar.getBeaconBlockRoot()); + } + @Test void getOrCreateTracker_shouldForceEvictOldestIncompleteAtHardLimit() { // hardLimit = maxRecentlySampledBlocks * 4 = 8 diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetrieverTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetrieverTest.java index ef19821783c..89925764988 100644 --- a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetrieverTest.java +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/retriever/SimpleSidecarRetrieverTest.java @@ -52,6 +52,7 @@ import tech.pegasys.teku.spec.logic.versions.fulu.helpers.MiscHelpersFulu; import tech.pegasys.teku.spec.util.DataStructureUtil; import tech.pegasys.teku.spec.util.KzgUtil; +import tech.pegasys.teku.statetransition.blobs.RemoteOrigin; import tech.pegasys.teku.statetransition.datacolumns.CanonicalBlockResolverStub; @SuppressWarnings({"JavaCase"}) @@ -311,6 +312,79 @@ void cancellingRequestShouldRemoveItFromPending() { assertThat(custodyPeer.getRequests()).hasSize(2); } + @Test + void sidecarReceivedFromGossipShouldCancelActiveRequest() { + final TestPeer custodyPeer = + new TestPeer(stubAsyncRunner, custodyNodeIds.next(), Duration.ofDays(1)); + testPeerManager.connectPeer(custodyPeer); + final DataColumnSidecar sidecar = createSidecarAndAddToAllPeers(1); + final SafeFuture result = + simpleSidecarRetriever.retrieve(DataColumnSlotAndIdentifier.fromDataColumn(sidecar)); + + advanceTimeGradually(retrieverRound); + assertThat(custodyPeer.getRequests()).hasSize(1); + assertThat(custodyPeer.getRequests().getFirst().response()).isNotDone(); + + simpleSidecarRetriever.onNewValidatedSidecar(sidecar, RemoteOrigin.GOSSIP); + advanceTimeGradually(Duration.ofMillis(1)); + + assertThat(result).isCompletedWithValue(sidecar); + assertThat(custodyPeer.getRequests().getFirst().response()).isCancelled(); + } + + @Test + void requestCancelledOnOurSideShouldNotAffectPeerScore() { + final TestPeer custodyPeer = + new TestPeer(stubAsyncRunner, custodyNodeIds.next(), Duration.ofDays(1)); + testPeerManager.connectPeer(custodyPeer); + final DataColumnSidecar sidecar = createSidecarAndAddToAllPeers(1, custodyPeer); + final SafeFuture result = + simpleSidecarRetriever.retrieve(DataColumnSlotAndIdentifier.fromDataColumn(sidecar)); + + advanceTimeGradually(retrieverRound); + final SimpleSidecarRetriever.ConnectedPeer connectedPeer = + simpleSidecarRetriever.getConnectedPeers().get(custodyPeer.getNodeId()); + assertThat(custodyPeer.getRequests()).hasSize(1); + assertThat(connectedPeer.getSidecarsRequested()).hasValue(2); + + // the sidecar arrives via gossip while the request is still in flight + simpleSidecarRetriever.onNewValidatedSidecar(sidecar, RemoteOrigin.GOSSIP); + advanceTimeGradually(Duration.ofMillis(1)); + + assertThat(result).isCompletedWithValue(sidecar); + assertThat(custodyPeer.getRequests().getFirst().response()).isCancelled(); + // we cancelled the request, so the peer shouldn't be penalised for not responding + assertThat(connectedPeer.getSidecarsRequested()).hasValue(1); + assertThat(connectedPeer.getResponseScore()).isEqualTo(10); + } + + @Test + void failedRequestShouldAffectPeerScore() { + final TestPeer custodyPeer = + new TestPeer(stubAsyncRunner, custodyNodeIds.next(), Duration.ofDays(1)); + testPeerManager.connectPeer(custodyPeer); + final DataColumnSidecar sidecar = createSidecarAndAddToAllPeers(1, custodyPeer); + simpleSidecarRetriever + .retrieve(DataColumnSlotAndIdentifier.fromDataColumn(sidecar)) + .finish(err -> LOG.trace("Error retrieving sidecar", err)); + + advanceTimeGradually(retrieverRound); + final SimpleSidecarRetriever.ConnectedPeer connectedPeer = + simpleSidecarRetriever.getConnectedPeers().get(custodyPeer.getNodeId()); + assertThat(custodyPeer.getRequests()).hasSize(1); + + custodyPeer + .getRequests() + .getFirst() + .response() + .completeExceptionally(new RuntimeException("Peer error")); + advanceTimeGradually(Duration.ofMillis(1)); + + // the peer failed to respond, so the request must still count against its score + assertThat(connectedPeer.getSidecarsRequested()).hasValue(2); + assertThat(connectedPeer.getResponseScore()).isEqualTo(5); + } + @Test @SuppressWarnings("unused") void performanceTest() { diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityCheckerTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityCheckerTest.java index 8fff7cea07d..a21858fa815 100644 --- a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityCheckerTest.java +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/DataColumnSidecarAvailabilityCheckerTest.java @@ -20,10 +20,12 @@ import java.time.Duration; import java.util.List; +import java.util.Optional; import java.util.concurrent.ExecutionException; import org.assertj.core.util.Lists; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import tech.pegasys.teku.infrastructure.async.DelayedExecutorAsyncRunner; import tech.pegasys.teku.infrastructure.async.SafeFuture; import tech.pegasys.teku.infrastructure.unsigned.UInt64; import tech.pegasys.teku.spec.Spec; @@ -47,10 +49,11 @@ class DataColumnSidecarAvailabilityCheckerTest { @BeforeEach void setup() { - checker = - new DataColumnSidecarAvailabilityChecker( - das, spec, recentChainData, block, Duration.ofSeconds(10)); + checker = new DataColumnSidecarAvailabilityChecker(das, spec, recentChainData, block); when(block.getMessage()).thenReturn(beaconBlock); + when(block.getSlot()).thenReturn(UInt64.ONE); + when(spec.getSlotDurationMillis(any())).thenReturn(10_000); + when(recentChainData.getCurrentSlot()).thenReturn(Optional.empty()); } @Test @@ -118,10 +121,10 @@ void shouldReturnNotAvailableOnTimeoutWhenBlockIsWithinDataAvailabilityWindow() when(das.checkDataAvailability(any(), any())).thenReturn(new SafeFuture<>()); when(recentChainData.getStore()).thenReturn(store); when(spec.isAvailabilityOfDataColumnSidecarsRequiredAtSlot(any(), any())).thenReturn(true); + when(spec.getSlotDurationMillis(any())).thenReturn(1); final DataColumnSidecarAvailabilityChecker timedOutChecker = - new DataColumnSidecarAvailabilityChecker( - das, spec, recentChainData, block, Duration.ofMillis(1)); + new DataColumnSidecarAvailabilityChecker(das, spec, recentChainData, block); assertThat(timedOutChecker.initiateDataAvailabilityCheck()).isTrue(); @@ -139,10 +142,10 @@ void shouldReturnNotRequiredOnTimeoutWhenBlockIsOutsideDataAvailabilityWindow() when(das.checkDataAvailability(any(), any())).thenReturn(new SafeFuture<>()); when(recentChainData.getStore()).thenReturn(store); when(spec.isAvailabilityOfDataColumnSidecarsRequiredAtSlot(any(), any())).thenReturn(false); + when(spec.getSlotDurationMillis(any())).thenReturn(1); final DataColumnSidecarAvailabilityChecker timedOutChecker = - new DataColumnSidecarAvailabilityChecker( - das, spec, recentChainData, block, Duration.ofMillis(1)); + new DataColumnSidecarAvailabilityChecker(das, spec, recentChainData, block); assertThat(timedOutChecker.initiateDataAvailabilityCheck()).isTrue(); @@ -160,10 +163,10 @@ void shouldNotPoisonSharedTrackerFutureOnTimeout() when(das.checkDataAvailability(any(), any())).thenReturn(sharedTrackerFuture); when(recentChainData.getStore()).thenReturn(store); when(spec.isAvailabilityOfDataColumnSidecarsRequiredAtSlot(any(), any())).thenReturn(true); + when(spec.getSlotDurationMillis(any())).thenReturn(1); final DataColumnSidecarAvailabilityChecker timedOutChecker = - new DataColumnSidecarAvailabilityChecker( - das, spec, recentChainData, block, Duration.ofMillis(1)); + new DataColumnSidecarAvailabilityChecker(das, spec, recentChainData, block); timedOutChecker.initiateDataAvailabilityCheck(); // wait for the checker to time out @@ -174,4 +177,46 @@ void shouldNotPoisonSharedTrackerFutureOnTimeout() // the shared tracker future must not have been completed by the timeout assertThat(sharedTrackerFuture).isNotDone(); } + + @Test + void shouldNotExtendTimeoutForBlockExactlyOneEpochOld() { + final UInt64 currentSlot = UInt64.valueOf(33); + final UpdatableStore store = mock(UpdatableStore.class); + when(recentChainData.getCurrentSlot()).thenReturn(Optional.of(currentSlot)); + when(recentChainData.getStore()).thenReturn(store); + when(spec.getSlotsPerEpoch(currentSlot)).thenReturn(32); + when(spec.getSlotDurationMillis(UInt64.ONE)).thenReturn(1); + when(spec.isAvailabilityOfDataColumnSidecarsRequiredAtSlot(store, UInt64.ONE)).thenReturn(true); + when(das.checkSamplingEligibility(block.getMessage())) + .thenReturn(DataAvailabilitySampler.SamplingEligibilityStatus.REQUIRED); + when(das.checkDataAvailability(UInt64.ONE, block.getRoot())).thenReturn(new SafeFuture<>()); + + assertThat(checker.initiateDataAvailabilityCheck()).isTrue(); + + assertThat(checker.getAvailabilityCheckResult()) + .succeedsWithin(Duration.ofSeconds(1)) + .satisfies(result -> assertThat(result.isNotAvailable()).isTrue()); + } + + @Test + void shouldExtendTimeoutForBlockOlderThanOneEpoch() { + final List sampledColumns = List.of(UInt64.ONE); + when(block.getSlot()).thenReturn(UInt64.ONE); + when(recentChainData.getCurrentSlot()).thenReturn(Optional.of(UInt64.valueOf(34))); + when(spec.getSlotsPerEpoch(UInt64.valueOf(34))).thenReturn(32); + when(spec.getSlotDurationMillis(UInt64.ONE)).thenReturn(100); + when(das.checkSamplingEligibility(block.getMessage())) + .thenReturn(DataAvailabilitySampler.SamplingEligibilityStatus.REQUIRED); + when(das.checkDataAvailability(UInt64.ONE, block.getRoot())) + .thenReturn( + DelayedExecutorAsyncRunner.create() + .runAfterDelay( + () -> SafeFuture.completedFuture(sampledColumns), Duration.ofMillis(250))); + + assertThat(checker.initiateDataAvailabilityCheck()).isTrue(); + + assertThat(checker.getAvailabilityCheckResult()) + .succeedsWithin(Duration.ofSeconds(1)) + .isEqualTo(DataAndValidationResult.validResult(sampledColumns)); + } } diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceTest.java index 28d163a5e28..2dc95856c55 100644 --- a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceTest.java +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceTest.java @@ -40,7 +40,6 @@ import it.unimi.dsi.fastutil.ints.IntList; import it.unimi.dsi.fastutil.ints.IntOpenHashSet; import it.unimi.dsi.fastutil.ints.IntSet; -import java.time.Duration; import java.util.List; import java.util.Optional; import java.util.Set; @@ -380,11 +379,12 @@ void onBlock_shouldFailWhenDataColumnAvailabilityNeverCompletes() throws Excepti when(dataAvailabilitySampler.checkDataAvailability(any(), any())) .thenReturn(new SafeFuture<>()); doReturn(true).when(spec).isAvailabilityOfDataColumnSidecarsRequiredAtSlot(any(), any()); + doReturn(1).when(spec).getSlotDurationMillis(any()); spec.reinitializeForTesting( block -> blobSidecarsAvailabilityChecker, block -> new DataColumnSidecarAvailabilityChecker( - dataAvailabilitySampler, spec, recentChainData, block, Duration.ofMillis(1)), + dataAvailabilitySampler, spec, recentChainData, block), KZG.DISABLED); final SignedBlockAndState blockAndState = chainBuilder.generateBlockAtSlot(ONE);