From 06227db7424d842f2628d27a2dc1aae8d3896c5e Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Thu, 20 Aug 2026 20:59:45 +0200 Subject: [PATCH 1/7] 1. Increase timeout, we have DA now 2. Cancel should cancel active batch import 3. Add debug logging --- .../sync/forward/multipeer/BatchImporter.java | 464 +++++++++++------- .../sync/forward/multipeer/BatchSync.java | 10 +- .../forward/multipeer/SyncStallDetector.java | 4 +- .../forward/multipeer/BatchImporterTest.java | 21 + .../sync/forward/multipeer/BatchSyncTest.java | 18 + 5 files changed, 328 insertions(+), 189 deletions(-) diff --git a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporter.java b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporter.java index 0292aadeeaa..e66bca8aec0 100644 --- a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporter.java +++ b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporter.java @@ -19,6 +19,8 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.CancellationException; +import java.util.concurrent.atomic.AtomicReference; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.tuweni.bytes.Bytes32; @@ -31,6 +33,7 @@ import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock; import tech.pegasys.teku.spec.datastructures.epbs.versions.gloas.SignedExecutionPayloadEnvelope; import tech.pegasys.teku.spec.logic.common.statetransition.results.BlockImportResult; +import tech.pegasys.teku.spec.logic.common.statetransition.results.ExecutionPayloadImportResult; import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool; import tech.pegasys.teku.statetransition.block.BlockImporter; import tech.pegasys.teku.statetransition.execution.ExecutionPayloadManager; @@ -65,126 +68,158 @@ public BatchImporter( */ public SafeFuture importBatch(final Batch batch) { // Copy the data from batch as we're going to use them from off the event thread. - final List blocks = new ArrayList<>(batch.getBlocks()); - final Map> blobSidecarsByBlockRoot = - Map.copyOf(batch.getBlobSidecarsByBlockRoot()); - final Map executionPayloadsByBlockRoot = - Map.copyOf(batch.getExecutionPayloadsByBlockRoot()); - - final Optional source = batch.getSource(); - - checkState(!blocks.isEmpty(), "Batch has no blocks to import"); - return asyncRunner.runAsync( - () -> { - final SignedBeaconBlock firstBlock = blocks.getFirst(); - SafeFuture importResult = - importBlock( - firstBlock, - blobSidecarsByBlockRoot, - executionPayloadsByBlockRoot, - source.orElseThrow()); - for (int i = 1; i < blocks.size(); i++) { - final SignedBeaconBlock block = blocks.get(i); - importResult = - importResult.thenCompose( - previousResult -> { - if (previousResult.isSuccessful()) { - return importBlock( - block, - blobSidecarsByBlockRoot, - executionPayloadsByBlockRoot, - source.orElseThrow()); - } else { - return SafeFuture.completedFuture(previousResult); - } - }); - } - return importResult.thenApply( - lastImportResult -> { - if (lastImportResult.isSuccessful()) { - return BatchImportResult.IMPORTED_ALL_BLOCKS; - } else if (lastImportResult.failedPayloadExecution()) { - return BatchImportResult.EXECUTION_CLIENT_OFFLINE; - } else if (lastImportResult.dataNotAvailable) { - return BatchImportResult.DATA_NOT_AVAILABLE; - } - LOG.debug( - "Failed to import batch {}: {}", - batch, - lastImportResult.failureReason(), - lastImportResult.failureCause().orElse(null)); - return BatchImportResult.IMPORT_FAILED; - }); - }); + final CancellableBatchImportFuture result = new CancellableBatchImportFuture(); + final BatchImportContext context = + new BatchImportContext( + batch, + new ArrayList<>(batch.getBlocks()), + Map.copyOf(batch.getBlobSidecarsByBlockRoot()), + Map.copyOf(batch.getExecutionPayloadsByBlockRoot()), + batch.getSource(), + result); + + checkState(!context.blocks().isEmpty(), "Batch has no blocks to import"); + asyncRunner + .runAsync(() -> importBlocks(context)) + .finish(result::complete, result::completeExceptionally); + return result; } - private SafeFuture importBlock( + private SafeFuture importBlocks(final BatchImportContext context) { + if (context.result().isCancelled()) { + return cancelledBatchImport(); + } + SafeFuture importResult = + importBlock(context, context.blocks().getFirst(), 1); + for (int i = 1; i < context.blocks().size(); i++) { + final SignedBeaconBlock block = context.blocks().get(i); + final int blockNumber = i + 1; + importResult = + importResult.thenCompose( + previousResult -> importNextBlock(context, block, blockNumber, previousResult)); + } + return importResult.thenApply( + lastImportResult -> toBatchImportResult(context.batch(), lastImportResult)); + } + + private SafeFuture importNextBlock( + final BatchImportContext context, final SignedBeaconBlock block, - final Map> blobSidecarsByBlockRoot, - final Map executionPayloadsByBlockRoot, - final SyncSource source) { + final int blockNumber, + final SingleImportResult previousResult) { + if (context.result().isCancelled()) { + return cancelledBatchImport(); + } + if (previousResult.isSuccessful()) { + return importBlock(context, block, blockNumber); + } + return SafeFuture.completedFuture(previousResult); + } + + private SafeFuture cancelledBatchImport() { + return SafeFuture.failedFuture(new CancellationException("Batch import cancelled")); + } + + private BatchImportResult toBatchImportResult( + final Batch batch, final SingleImportResult lastImportResult) { + if (lastImportResult.isSuccessful()) { + return BatchImportResult.IMPORTED_ALL_BLOCKS; + } else if (lastImportResult.failedPayloadExecution()) { + return BatchImportResult.EXECUTION_CLIENT_OFFLINE; + } else if (lastImportResult.dataNotAvailable) { + return BatchImportResult.DATA_NOT_AVAILABLE; + } + LOG.debug( + "Failed to import batch {}: {}", + batch, + lastImportResult.failureReason(), + lastImportResult.failureCause().orElse(null)); + return BatchImportResult.IMPORT_FAILED; + } + + private SafeFuture importBlock( + final BatchImportContext context, final SignedBeaconBlock block, final int blockNumber) { + LOG.debug( + "Importing block {} of {} during syncing for slot {} and root {}", + blockNumber, + context.blocks().size(), + block.getSlot(), + block.getRoot()); final Bytes32 blockRoot = block.getRoot(); final Optional executionPayload = - Optional.ofNullable(executionPayloadsByBlockRoot.get(blockRoot)); - if (!blobSidecarsByBlockRoot.containsKey(blockRoot)) { - return importBlock(block, executionPayload, source); + Optional.ofNullable(context.executionPayloadsByBlockRoot().get(blockRoot)); + if (context.blobSidecarsByBlockRoot().containsKey(blockRoot)) { + final List blobSidecars = context.blobSidecarsByBlockRoot().get(blockRoot); + LOG.trace( + "Sending {} blob sidecars to the pool during syncing for block with root {}", + blobSidecars.size(), + blockRoot); + // Add blob sidecars to the pool in order for them to be available when the block is being + // imported + blockBlobSidecarsTrackersPool.onCompletedBlockAndBlobSidecars(block, blobSidecars); } - final List blobSidecars = blobSidecarsByBlockRoot.get(blockRoot); - LOG.trace( - "Sending {} blob sidecars to the pool during syncing for block with root {}", - blobSidecars.size(), - blockRoot); - // Add blob sidecars to the pool in order for them to be available when the block is being - // imported - blockBlobSidecarsTrackersPool.onCompletedBlockAndBlobSidecars(block, blobSidecars); - return importBlock(block, executionPayload, source); + + return importBlockWithParentExecutionPayloadRecovery( + block, context.source().orElseThrow(), context.result()) + .thenCompose( + blockImportResult -> + handleBlockImportResult( + block, + executionPayload, + context.source.orElseThrow(), + context.result(), + blockImportResult)); } - private SafeFuture importBlock( + private SafeFuture handleBlockImportResult( final SignedBeaconBlock block, final Optional executionPayload, - final SyncSource source) { + final SyncSource source, + final CancellableBatchImportFuture batchImportFuture, + final BlockImportResult blockImportResult) { + if (blockImportResult.getFailureReason() + == BlockImportResult.FailureReason.FAILED_WEAK_SUBJECTIVITY_CHECKS) { + LOG.warn( + "Disconnecting source ({}) for sending block that failed weak subjectivity checks: {}", + source, + blockImportResult); + source.disconnectCleanly(DisconnectReason.REMOTE_FAULT).finishWarn(LOG); + } + if (executionPayload.isEmpty() || !blockImportResult.isSuccessful()) { + return SafeFuture.completedFuture(toSingleImportResult(blockImportResult)); + } LOG.trace( - "Importing block during syncing for slot {} and root {}", block.getSlot(), block.getRoot()); - return importBlockWithParentExecutionPayloadRecovery(block, source) - .thenCompose( - blockImportResult -> { - if (blockImportResult.getFailureReason() - == BlockImportResult.FailureReason.FAILED_WEAK_SUBJECTIVITY_CHECKS) { - LOG.warn( - "Disconnecting source ({}) for sending block that failed weak subjectivity checks: {}", - source, - blockImportResult); - source.disconnectCleanly(DisconnectReason.REMOTE_FAULT).finishWarn(LOG); - } - if (executionPayload.isEmpty() || !blockImportResult.isSuccessful()) { - return SafeFuture.completedFuture( - new SingleImportResult( - blockImportResult.isSuccessful(), - blockImportResult.hasFailedExecutingExecutionPayload(), - blockImportResult.isDataNotAvailable(), - Optional.ofNullable(blockImportResult.getFailureReason()) - .map(Enum::name) - .orElse(null), - blockImportResult.getFailureCause())); - } - LOG.trace( - "Importing execution payload during syncing for slot {} and block root {}", - block.getSlot(), - block.getRoot()); - return executionPayloadManager - .importExecutionPayload(executionPayload.get(), false) - .thenApply( - executionPayloadImportResult -> - new SingleImportResult( - executionPayloadImportResult.isSuccessful(), - executionPayloadImportResult.hasFailedExecution(), - executionPayloadImportResult.isDataNotAvailable(), - Optional.ofNullable(executionPayloadImportResult.getFailureReason()) - .map(Enum::name) - .orElse(null), - executionPayloadImportResult.getFailureCause())); - }); + "Importing execution payload during syncing for slot {} and block root {}", + block.getSlot(), + block.getRoot()); + return trackActiveTask( + executionPayloadManager.importExecutionPayload(executionPayload.get(), false), + batchImportFuture) + .thenApply(this::toSingleImportResult); + } + + private SingleImportResult toSingleImportResult(final BlockImportResult result) { + return new SingleImportResult( + result.isSuccessful(), + result.hasFailedExecutingExecutionPayload(), + result.isDataNotAvailable(), + Optional.ofNullable(result.getFailureReason()).map(Enum::name).orElse(null), + result.getFailureCause()); + } + + private SingleImportResult toSingleImportResult(final ExecutionPayloadImportResult result) { + return new SingleImportResult( + result.isSuccessful(), + result.hasFailedExecution(), + result.isDataNotAvailable(), + Optional.ofNullable(result.getFailureReason()).map(Enum::name).orElse(null), + result.getFailureCause()); + } + + private SafeFuture trackActiveTask( + final SafeFuture task, final CancellableBatchImportFuture batchImportFuture) { + batchImportFuture.setActiveTask(task); + return task; } /** @@ -199,94 +234,153 @@ private SafeFuture importBlock( * data unavailable, in which case that failure is forwarded to the batch result. */ private SafeFuture importBlockWithParentExecutionPayloadRecovery( - final SignedBeaconBlock block, final SyncSource source) { - return blockImporter - .importBlock(block) + final SignedBeaconBlock block, + final SyncSource source, + final CancellableBatchImportFuture batchImportFuture) { + return trackActiveTask(blockImporter.importBlock(block), batchImportFuture) .thenCompose( - result -> { - if (result.getFailureReason() - != BlockImportResult.FailureReason.UNKNOWN_PARENT_EXECUTION_PAYLOAD) { - return SafeFuture.completedFuture(result); - } - LOG.debug( - "Recovering missing parent execution payload by root {} for block at slot {}", - block.getParentRoot(), - block.getSlot()); - return recoverParentExecutionPayloadByRoot(block, source, result); - }); + result -> + recoverParentExecutionPayloadIfRequired(block, source, batchImportFuture, result)); + } + + private SafeFuture recoverParentExecutionPayloadIfRequired( + final SignedBeaconBlock block, + final SyncSource source, + final CancellableBatchImportFuture batchImportFuture, + final BlockImportResult result) { + if (result.getFailureReason() + != BlockImportResult.FailureReason.UNKNOWN_PARENT_EXECUTION_PAYLOAD) { + return SafeFuture.completedFuture(result); + } + LOG.debug( + "Recovering missing parent execution payload by root {} for block at slot {}", + block.getParentRoot(), + block.getSlot()); + return recoverParentExecutionPayloadByRoot(block, source, result, batchImportFuture); } private SafeFuture recoverParentExecutionPayloadByRoot( - final SignedBeaconBlock block, final SyncSource source, final BlockImportResult result) { - return source - .requestExecutionPayloadEnvelopeByRoot(block.getParentRoot()) + final SignedBeaconBlock block, + final SyncSource source, + final BlockImportResult result, + final CancellableBatchImportFuture batchImportFuture) { + return trackActiveTask( + source.requestExecutionPayloadEnvelopeByRoot(block.getParentRoot()), batchImportFuture) .thenCompose( maybeExecutionPayload -> - maybeExecutionPayload - .map( - signedExecutionPayloadEnvelope -> - importRecoveredParentExecutionPayload( - block, signedExecutionPayloadEnvelope, result)) - .orElseGet( - () -> { - LOG.debug( - "Failed to recover parent execution payload by root {} for block at slot {}: no envelope returned", - block.getParentRoot(), - block.getSlot()); - return SafeFuture.completedFuture(result); - })) - .exceptionally( - error -> { - LOG.debug( - "Failed to recover parent execution payload by root {} for block at slot {}", - block.getParentRoot(), - block.getSlot(), - error); - return result; - }); + handleParentExecutionPayloadResponse( + block, result, batchImportFuture, maybeExecutionPayload)) + .exceptionally(error -> handleParentExecutionPayloadRecoveryError(block, result, error)); + } + + private SafeFuture handleParentExecutionPayloadResponse( + final SignedBeaconBlock block, + final BlockImportResult result, + final CancellableBatchImportFuture batchImportFuture, + final Optional maybeExecutionPayload) { + if (maybeExecutionPayload.isPresent()) { + return importRecoveredParentExecutionPayload( + block, maybeExecutionPayload.get(), result, batchImportFuture); + } + LOG.debug( + "Failed to recover parent execution payload by root {} for block at slot {}: no envelope returned", + block.getParentRoot(), + block.getSlot()); + return SafeFuture.completedFuture(result); + } + + private BlockImportResult handleParentExecutionPayloadRecoveryError( + final SignedBeaconBlock block, final BlockImportResult result, final Throwable error) { + LOG.debug( + "Failed to recover parent execution payload by root {} for block at slot {}", + block.getParentRoot(), + block.getSlot(), + error); + return result; } private SafeFuture importRecoveredParentExecutionPayload( final SignedBeaconBlock block, final SignedExecutionPayloadEnvelope signedExecutionPayloadEnvelope, - final BlockImportResult originalResult) { - return executionPayloadManager - .importExecutionPayload(signedExecutionPayloadEnvelope, false) + final BlockImportResult originalResult, + final CancellableBatchImportFuture batchImportFuture) { + return trackActiveTask( + executionPayloadManager.importExecutionPayload(signedExecutionPayloadEnvelope, false), + batchImportFuture) .thenCompose( - executionPayloadImportResult -> { - if (executionPayloadImportResult.isSuccessful()) { - return blockImporter.importBlock(block); - } - if (executionPayloadImportResult.hasFailedExecution()) { - LOG.debug( - "Failed to import recovered parent execution payload by root {} for block at slot {}: {}", - block.getParentRoot(), - block.getSlot(), - executionPayloadImportResult.toLogString(), - executionPayloadImportResult.getFailureCause().orElse(null)); - return SafeFuture.completedFuture( - BlockImportResult.failedExecutionPayloadExecution( - executionPayloadImportResult.getFailureCause().orElseThrow())); - } - if (executionPayloadImportResult.isDataNotAvailable()) { - LOG.debug( - "Recovered parent execution payload by root {} for block at slot {} is data unavailable: {}", - block.getParentRoot(), - block.getSlot(), - executionPayloadImportResult.toLogString(), - executionPayloadImportResult.getFailureCause().orElse(null)); - return SafeFuture.completedFuture( - BlockImportResult.failedDataAvailabilityCheckNotAvailable( - executionPayloadImportResult.getFailureCause())); - } - LOG.debug( - "Failed to import recovered parent execution payload by root {} for block at slot {}: {}", - block.getParentRoot(), - block.getSlot(), - executionPayloadImportResult.toLogString(), - executionPayloadImportResult.getFailureCause().orElse(null)); - return SafeFuture.completedFuture(originalResult); - }); + executionPayloadImportResult -> + handleRecoveredParentExecutionPayloadImport( + block, originalResult, batchImportFuture, executionPayloadImportResult)); + } + + private SafeFuture handleRecoveredParentExecutionPayloadImport( + final SignedBeaconBlock block, + final BlockImportResult originalResult, + final CancellableBatchImportFuture batchImportFuture, + final ExecutionPayloadImportResult executionPayloadImportResult) { + if (executionPayloadImportResult.isSuccessful()) { + return trackActiveTask(blockImporter.importBlock(block), batchImportFuture); + } + if (executionPayloadImportResult.hasFailedExecution()) { + LOG.debug( + "Failed to import recovered parent execution payload by root {} for block at slot {}: {}", + block.getParentRoot(), + block.getSlot(), + executionPayloadImportResult.toLogString(), + executionPayloadImportResult.getFailureCause().orElse(null)); + return SafeFuture.completedFuture( + BlockImportResult.failedExecutionPayloadExecution( + executionPayloadImportResult.getFailureCause().orElseThrow())); + } + if (executionPayloadImportResult.isDataNotAvailable()) { + LOG.debug( + "Recovered parent execution payload by root {} for block at slot {} is data unavailable: {}", + block.getParentRoot(), + block.getSlot(), + executionPayloadImportResult.toLogString(), + executionPayloadImportResult.getFailureCause().orElse(null)); + return SafeFuture.completedFuture( + BlockImportResult.failedDataAvailabilityCheckNotAvailable( + executionPayloadImportResult.getFailureCause())); + } + LOG.debug( + "Failed to import recovered parent execution payload by root {} for block at slot {}: {}", + block.getParentRoot(), + block.getSlot(), + executionPayloadImportResult.toLogString(), + executionPayloadImportResult.getFailureCause().orElse(null)); + return SafeFuture.completedFuture(originalResult); + } + + private record BatchImportContext( + Batch batch, + List blocks, + Map> blobSidecarsByBlockRoot, + Map executionPayloadsByBlockRoot, + Optional source, + CancellableBatchImportFuture result) {} + + private static class CancellableBatchImportFuture extends SafeFuture { + private final AtomicReference> activeTask = new AtomicReference<>(); + + private void setActiveTask(final SafeFuture task) { + activeTask.set(task); + if (isCancelled()) { + task.cancel(true); + } + } + + @Override + public boolean cancel(final boolean mayInterruptIfRunning) { + final boolean cancelled = super.cancel(mayInterruptIfRunning); + if (cancelled) { + final SafeFuture task = activeTask.get(); + if (task != null) { + task.cancel(mayInterruptIfRunning); + } + } + return cancelled; + } } public record SingleImportResult( 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 2b54102f220..db71ce0654e 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 @@ -59,6 +59,7 @@ public class BatchSync implements Sync { private final BatchChain activeBatches; private Optional importingBatch = Optional.empty(); + private Optional> importingBatchFuture = Optional.empty(); private boolean switchingBranches = false; private SafeFuture commonAncestorSlot; @@ -387,8 +388,9 @@ private void startNextImport() { batch -> { lastImportTimerStartPointSeconds = timeProvider.getTimeInSeconds(); importingBatch = Optional.of(batch); - batchImporter - .importBatch(batch) + final SafeFuture importFuture = batchImporter.importBatch(batch); + importingBatchFuture = Optional.of(importFuture); + importFuture .thenAcceptAsync(result -> onImportComplete(result, batch), eventThread) .propagateExceptionTo(syncResult); }); @@ -421,6 +423,7 @@ private void onImportComplete( isCurrentlyImportingBatch(importedBatch), "Received import complete for batch that shouldn't have been importing"); importingBatch = Optional.empty(); + importingBatchFuture = Optional.empty(); if (switchingBranches) { // We switched to a different chain while this was importing. Can't infer anything about other // batches from this result but should still penalise the peer that sent it to us. @@ -558,12 +561,15 @@ Optional getImportingBatch() { public void abort() { eventThread.checkOnEventThread(); LOG.warn("Aborting sync {}", this::describeState); + final Optional> importToCancel = importingBatchFuture; importingBatch = Optional.empty(); + importingBatchFuture = Optional.empty(); activeBatches.removeAll(); switchingBranches = false; commonAncestorSlot = null; targetChain = null; syncResult.complete(SyncResult.FAILED); + importToCancel.ifPresent(importFuture -> importFuture.cancel(true)); } private String describeState() { diff --git a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/SyncStallDetector.java b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/SyncStallDetector.java index 8dc270b8a63..1833f3acbb0 100644 --- a/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/SyncStallDetector.java +++ b/beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/SyncStallDetector.java @@ -37,8 +37,8 @@ public class SyncStallDetector extends Service { static final Duration STALL_CHECK_INTERVAL = Duration.ofSeconds(15); // Time periods are fairly long because sync stalls should be rare and we might be rate limited // if we have to request blocks from a small number of peers. - static final int MAX_SECONDS_BETWEEN_IMPORTS = 180; - static final int MAX_SECONDS_BETWEEN_IMPORT_PROGRESS = 180; + static final int MAX_SECONDS_BETWEEN_IMPORTS = 300; + static final int MAX_SECONDS_BETWEEN_IMPORT_PROGRESS = 300; private final Spec spec; private final EventThread eventThread; diff --git a/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporterTest.java b/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporterTest.java index 1a29d246522..751d30ab6e9 100644 --- a/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporterTest.java +++ b/beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchImporterTest.java @@ -115,6 +115,27 @@ void shouldImportBlocksInOrder() { verifyNoMoreInteractions(batch); } + @Test + void shouldCancelActiveBlockImport() { + final SignedBeaconBlock block1 = dataStructureUtil.randomSignedBeaconBlock(1); + final SignedBeaconBlock block2 = dataStructureUtil.randomSignedBeaconBlock(2); + final SafeFuture importResult1 = new SafeFuture<>(); + final SafeFuture importResult2 = new SafeFuture<>(); + when(batch.getBlocks()).thenReturn(List.of(block1, block2)); + when(blockImporter.importBlock(block1)).thenReturn(importResult1); + when(blockImporter.importBlock(block2)).thenReturn(importResult2); + + final SafeFuture result = importer.importBatch(batch); + asyncRunner.executeQueuedActions(); + + ignoreFuture(verify(blockImporter).importBlock(block1)); + assertThat(result.cancel(true)).isTrue(); + + assertThat(result).isCancelled(); + assertThat(importResult1).isCancelled(); + verify(blockImporter, never()).importBlock(block2); + } + @Test void shouldImportBlobSidecarsAndBlocksInOrder() { final SignedBeaconBlock block1 = dataStructureUtil.randomSignedBeaconBlock(1); 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 a61dbf64ea2..e7019088c6f 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 @@ -36,6 +36,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import tech.pegasys.teku.beacon.sync.events.SyncPreImportBlockChannel; +import tech.pegasys.teku.beacon.sync.forward.multipeer.BatchImporter.BatchImportResult; import tech.pegasys.teku.beacon.sync.forward.multipeer.Sync.SyncProgress; import tech.pegasys.teku.beacon.sync.forward.multipeer.batches.Batch; import tech.pegasys.teku.beacon.sync.forward.multipeer.batches.StubBatchFactory; @@ -152,6 +153,23 @@ void shouldImportFirstBatchWhenSecondBatchFormsChain() { assertBatchImported(batch1); } + @Test + void shouldCancelInProgressImportWhenAborted() { + final SignedBlockAndState block5 = chainBuilder.generateBlockAtSlot(5); + final SignedBlockAndState block26 = chainBuilder.generateBlockAtSlot(26); + final SafeFuture syncResult = sync.syncToChain(targetChain); + + final Batch batch1 = batches.get(0); + batches.receiveBlocks(batch1, block5.getBlock()); + batches.receiveBlocks(batches.get(1), block26.getBlock()); + final SafeFuture importResult = batches.getImportResult(batch1); + + eventThread.execute(sync::abort); + + assertThat(syncResult).isCompletedWithValue(SyncResult.FAILED); + assertThat(importResult).isCancelled(); + } + @Test void shouldNotImportPreviousBatchWhenBoundaryParentExecutionPayloadIsMissing() { final UInt64 gloasForkEpoch = UInt64.valueOf(1000); From 9ce50da29eb1d1cc034002bf90dd9374ef88ca94 Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Mon, 24 Aug 2026 20:44:35 +0200 Subject: [PATCH 2/7] just some guards and cleaning around BatchSync --- .../pegasys/teku/beacon/sync/SyncConfig.java | 4 +-- .../sync/forward/multipeer/BatchSync.java | 6 ++++- .../datacolumns/DasSamplerBasicImpl.java | 17 ++++++++---- .../retriever/SimpleSidecarRetriever.java | 26 ++++++++++++++++--- .../datacolumns/DasSamplerBasicTest.java | 26 +++++++++++++++++++ .../retriever/SimpleSidecarRetrieverTest.java | 21 +++++++++++++++ 6 files changed, 89 insertions(+), 11 deletions(-) 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..65df20cf34d 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 @@ -448,7 +448,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/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..527e131feff 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 @@ -140,6 +140,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; @@ -169,10 +173,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 +251,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)) { + request.activeRpcRequest = null; + request.activeRpcRequestSet.set(false); request.result.completeAsync(maybeResult, asyncRunner); + final ActiveRequest activeRequest = request.activeRpcRequest; + if (activeRequest != null) { + activeRequest.promise().cancel(true); + } retrieveCounter.incrementAndGet(); } else if (request.activeRpcRequestSet.compareAndSet(true, false)) { request.activeRpcRequest = null; @@ -300,7 +320,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; 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..7d5cd439ca8 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 @@ -569,6 +569,7 @@ void onBlockImported_shouldRetainTrackerWhenDataAvailabilityDeferredToExecutionP assertThat(gloasSampler.getRecentlySampledColumnsByRoot()) .containsKey(blockWithBlobs.getRoot()); + assertThat(gloasSampler.getBlock(blockWithBlobs.getRoot())).isEmpty(); gloasSampler.onExecutionPayloadImported(blockWithBlobs.getSlotAndBlockRoot()); @@ -691,6 +692,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 +724,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 +741,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..01e3390bed1 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,26 @@ 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 @SuppressWarnings("unused") void performanceTest() { From 81dd80b0f8cac71dfbe1187c698c008189caebb5 Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Mon, 24 Aug 2026 23:12:20 +0200 Subject: [PATCH 3/7] fix incorrect changes --- .../datacolumns/retriever/SimpleSidecarRetriever.java | 2 +- .../teku/statetransition/datacolumns/DasSamplerBasicTest.java | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) 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 527e131feff..fc45d054b2a 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 @@ -252,10 +252,10 @@ private void nextRound() { private void reqRespCompleted( final RetrieveRequest request, final DataColumnSidecar maybeResult) { 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); - final ActiveRequest activeRequest = request.activeRpcRequest; if (activeRequest != null) { activeRequest.promise().cancel(true); } 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 7d5cd439ca8..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 @@ -569,7 +569,6 @@ void onBlockImported_shouldRetainTrackerWhenDataAvailabilityDeferredToExecutionP assertThat(gloasSampler.getRecentlySampledColumnsByRoot()) .containsKey(blockWithBlobs.getRoot()); - assertThat(gloasSampler.getBlock(blockWithBlobs.getRoot())).isEmpty(); gloasSampler.onExecutionPayloadImported(blockWithBlobs.getSlotAndBlockRoot()); From 29f2463c31948101ab11025129881b250172f8d0 Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Mon, 24 Aug 2026 23:12:29 +0200 Subject: [PATCH 4/7] increase timeout for older slots --- .../DataColumnSidecarAvailabilityChecker.java | 32 ++++------ ...aColumnSidecarAvailabilityCheckerTest.java | 63 ++++++++++++++++--- .../forkchoice/ForkChoiceTest.java | 4 +- 3 files changed, 68 insertions(+), 31 deletions(-) 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/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 0125d864218..6692cce4eef 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); From 413023706a5b909d1683104f4d5207c361009fcf Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Wed, 26 Aug 2026 11:21:46 +0200 Subject: [PATCH 5/7] Fix: failed request affected peer score --- .../retriever/SimpleSidecarRetriever.java | 17 ++++++ .../retriever/SimpleSidecarRetrieverTest.java | 53 +++++++++++++++++++ 2 files changed, 70 insertions(+) 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 fc45d054b2a..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; @@ -159,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: {}", @@ -393,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/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 01e3390bed1..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 @@ -332,6 +332,59 @@ void sidecarReceivedFromGossipShouldCancelActiveRequest() { 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() { From 2f15df0f0ef53aaa4e89698372b80f78354d6091 Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Wed, 26 Aug 2026 12:04:38 +0200 Subject: [PATCH 6/7] Fix: sync stall when import complete exceptionally --- .../sync/forward/multipeer/BatchSync.java | 34 +++++++++++++++++-- .../sync/forward/multipeer/BatchSyncTest.java | 28 +++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) 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 65df20cf34d..f2bf317ac7b 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 @@ -390,9 +390,22 @@ private void startNextImport() { importingBatch = Optional.of(batch); final SafeFuture importFuture = batchImporter.importBatch(batch); importingBatchFuture = Optional.of(importFuture); + final SafeFuture currentSyncResult = syncResult; importFuture - .thenAcceptAsync(result -> onImportComplete(result, batch), eventThread) - .propagateExceptionTo(syncResult); + .handleAsync( + (result, error) -> { + if (error != null) { + // clear the importing state before failing the sync, otherwise a + // restarted sync would wait forever for an import that already finished + onImportFailed(batch, error); + currentSyncResult.completeExceptionally(error); + } else { + onImportComplete(result, batch); + } + return null; + }, + eventThread) + .propagateExceptionTo(currentSyncResult); }); } @@ -416,6 +429,23 @@ private void markBatchesAsContested(final NavigableSet contestedBatches) contestedBatches.forEach(Batch::markAsContested); } + /** + * Handles an import which completed exceptionally (for example the import was cancelled or failed + * unexpectedly). Nothing can be inferred about the batch itself, but the importing state has to + * be released so that a later sync isn't blocked waiting for this import to complete. + */ + 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; + } + LOG.debug("Import of batch {} failed", importedBatch, error); + importingBatch = Optional.empty(); + importingBatchFuture = Optional.empty(); + switchingBranches = false; + } + private void onImportComplete( final BatchImporter.BatchImportResult result, final Batch importedBatch) { eventThread.checkOnEventThread(); 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..6f7a8528c77 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,33 @@ void shouldCancelInProgressImportWhenAborted() { assertThat(importResult).isCancelled(); } + @Test + void shouldReleaseImportingBatchWhenImportCompletesExceptionally() { + final SignedBlockAndState block5 = chainBuilder.generateBlockAtSlot(5); + final SignedBlockAndState block26 = chainBuilder.generateBlockAtSlot(26); + final SafeFuture syncResult = sync.syncToChain(targetChain); + + final Batch batch1 = batches.get(0); + batches.receiveBlocks(batch1, block5.getBlock()); + batches.receiveBlocks(batches.get(1), block26.getBlock()); + final SafeFuture importResult = batches.getImportResult(batch1); + + final RuntimeException importError = new RuntimeException("Import blew up"); + importResult.completeExceptionally(importError); + + SafeFutureAssert.assertThatSafeFuture(syncResult).isCompletedExceptionallyWith(importError); + + // The failed import must not block the next sync + final int batchCountBeforeRestart = batches.size(); + assertThat(sync.syncToChain(targetChain)).isNotDone(); + + // The failed batch is dropped and downloading restarts from the common ancestor + verify(commonAncestor, times(2)).findCommonAncestor(any()); + assertBatchNotActive(batch1); + assertThat(batches.size()).isGreaterThan(batchCountBeforeRestart); + assertThatBatch(batches.get(batchCountBeforeRestart)).hasFirstSlot(ONE); + } + @Test void shouldNotImportPreviousBatchWhenBoundaryParentExecutionPayloadIsMissing() { final UInt64 gloasForkEpoch = UInt64.valueOf(1000); From f4e4c6bfc40ec11b99dea8b58b3aaf7812d9e66a Mon Sep 17 00:00:00 2001 From: Dmitrii Shmatko Date: Wed, 26 Aug 2026 14:12:19 +0200 Subject: [PATCH 7/7] Fix: releasing exceptionally failed import --- .../sync/forward/multipeer/BatchSync.java | 20 +++--- .../sync/forward/multipeer/BatchSyncTest.java | 70 ++++++++++++++----- 2 files changed, 61 insertions(+), 29 deletions(-) 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 f2bf317ac7b..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 @@ -390,22 +390,18 @@ private void startNextImport() { importingBatch = Optional.of(batch); final SafeFuture importFuture = batchImporter.importBatch(batch); importingBatchFuture = Optional.of(importFuture); - final SafeFuture currentSyncResult = syncResult; importFuture .handleAsync( (result, error) -> { if (error != null) { - // clear the importing state before failing the sync, otherwise a - // restarted sync would wait forever for an import that already finished onImportFailed(batch, error); - currentSyncResult.completeExceptionally(error); } else { onImportComplete(result, batch); } return null; }, eventThread) - .propagateExceptionTo(currentSyncResult); + .propagateExceptionTo(syncResult); }); } @@ -430,9 +426,10 @@ private void markBatchesAsContested(final NavigableSet contestedBatches) } /** - * Handles an import which completed exceptionally (for example the import was cancelled or failed - * unexpectedly). Nothing can be inferred about the batch itself, but the importing state has to - * be released so that a later sync isn't blocked waiting for this import to complete. + * 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(); @@ -440,10 +437,9 @@ private void onImportFailed(final Batch importedBatch, final Throwable error) { // already released, e.g. the sync was aborted which cancelled the import return; } - LOG.debug("Import of batch {} failed", importedBatch, error); - importingBatch = Optional.empty(); - importingBatchFuture = Optional.empty(); - switchingBranches = false; + // 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( 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 6f7a8528c77..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 @@ -172,30 +172,28 @@ void shouldCancelInProgressImportWhenAborted() { } @Test - void shouldReleaseImportingBatchWhenImportCompletesExceptionally() { + void shouldHandleImportCompletingExceptionallyAsFailedImport() { final SignedBlockAndState block5 = chainBuilder.generateBlockAtSlot(5); final SignedBlockAndState block26 = chainBuilder.generateBlockAtSlot(26); final SafeFuture syncResult = sync.syncToChain(targetChain); - final Batch batch1 = batches.get(0); - batches.receiveBlocks(batch1, block5.getBlock()); - batches.receiveBlocks(batches.get(1), block26.getBlock()); - final SafeFuture importResult = batches.getImportResult(batch1); - - final RuntimeException importError = new RuntimeException("Import blew up"); - importResult.completeExceptionally(importError); + 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); - SafeFutureAssert.assertThatSafeFuture(syncResult).isCompletedExceptionallyWith(importError); + importResult.completeExceptionally(new RuntimeException("Import blew up")); - // The failed import must not block the next sync - final int batchCountBeforeRestart = batches.size(); - assertThat(sync.syncToChain(targetChain)).isNotDone(); + // 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(); - // The failed batch is dropped and downloading restarts from the common ancestor - verify(commonAncestor, times(2)).findCommonAncestor(any()); - assertBatchNotActive(batch1); - assertThat(batches.size()).isGreaterThan(batchCountBeforeRestart); - assertThatBatch(batches.get(batchCountBeforeRestart)).hasFirstSlot(ONE); + // 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 @@ -972,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();