Other sync improvements

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;
 

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/BatchSync.java

@@ -391,7 +391,16 @@ private void startNextImport() {
               final SafeFuture<BatchImportResult> 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<Batch> 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(
                 () ->

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> 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<BatchImportResult> 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<SyncResult> 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();

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<DataColumnSidecar> retrieveColumnWithSamplingAndCustody(
       final DataColumnSlotAndIdentifier id, final DataColumnSamplingTracker tracker) {
     return retriever

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<RequestMatch> 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<ConnectedPeer> findBestMatchingPeer(
       final RetrieveRequest request, final RequestTracker ongoingRequestsTracker) {
     final Stream<ConnectedPeer> 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<UInt256, ConnectedPeer> getConnectedPeers() {
     return connectedPeers;
   }
 
-  private record ActiveRequest(SafeFuture<Void> 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);

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<SignedExecutionPayloadEnvelope> 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));
   }
 }

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