SingleBockProviderFulu with soft and hard limit for sampled blocks

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/DefaultSyncServiceFactory.java

@@ -43,6 +43,7 @@
 import tech.pegasys.teku.statetransition.blobs.BlobSidecarManager;
 import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool;
 import tech.pegasys.teku.statetransition.block.BlockImporter;
+import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
 import tech.pegasys.teku.statetransition.execution.ExecutionPayloadManager;
 import tech.pegasys.teku.statetransition.util.PendingPool;
 import tech.pegasys.teku.statetransition.validation.signatures.SignatureVerificationService;
@@ -75,6 +76,7 @@ public class DefaultSyncServiceFactory implements SyncServiceFactory {
   private final PendingPool<SignedBeaconBlock> pendingBlocks;
   private final PendingPool<ValidatableAttestation> pendingAttestations;
   private final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool;
+  private final DasSamplerBasic dasSamplerBasic;
   private final int getStartupTargetPeerCount;
   private final AsyncBLSSignatureVerifier signatureVerifier;
   private final Duration startupTimeout;
@@ -98,6 +100,7 @@ public DefaultSyncServiceFactory(
       final PendingPool<SignedBeaconBlock> pendingBlocks,
       final PendingPool<ValidatableAttestation> pendingAttestations,
       final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool,
+      final DasSamplerBasic dasSamplerBasic,
       final int getStartupTargetPeerCount,
       final SignatureVerificationService signatureVerifier,
       final Duration startupTimeout,
@@ -119,6 +122,7 @@ public DefaultSyncServiceFactory(
     this.pendingBlocks = pendingBlocks;
     this.pendingAttestations = pendingAttestations;
     this.blockBlobSidecarsTrackersPool = blockBlobSidecarsTrackersPool;
+    this.dasSamplerBasic = dasSamplerBasic;
     this.getStartupTargetPeerCount = getStartupTargetPeerCount;
     this.signatureVerifier = signatureVerifier;
     this.startupTimeout = startupTimeout;
@@ -141,6 +145,7 @@ public SyncService create(final EventChannels eventChannels) {
             pendingBlocks,
             pendingAttestations,
             blockBlobSidecarsTrackersPool,
+            dasSamplerBasic,
             forwardSyncService,
             fetchTaskFactory);
     final RecentBlobSidecarsFetcher recentBlobSidecarsFetcher =

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/SyncConfig.java

@@ -28,7 +28,12 @@ public class SyncConfig {
   public static final int DEFAULT_FORWARD_SYNC_BATCH_SIZE = 25;
   public static final int DEFAULT_FORWARD_SYNC_MAX_PENDING_BATCHES = 5;
 
-  /** Aligned with {@link P2PConfig#DEFAULT_PEER_BLOCKS_RATE_LIMIT} */
+  /**
+   * Must be >= FORWARD_SYNC_BATCH_SIZE * FORWARD_SYNC_MAX_PENDING_BATCHES to avoid evicting
+   * completed trackers before the sync pipeline imports them.
+   */
+  public static final int DEFAULT_MAX_RECENTLY_SAMPLED_BLOCKS = 128;
+
   public static final int DEFAULT_FORWARD_SYNC_MAX_BLOCKS_PER_MINUTE = 500;
 
   /** Aligned with {@link P2PConfig#DEFAULT_PEER_BLOB_SIDECARS_RATE_LIMIT} */
@@ -43,6 +48,7 @@ public class SyncConfig {
   private final int forwardSyncMaxPendingBatches;
   private final int forwardSyncMaxBlocksPerMinute;
   private final int forwardSyncMaxBlobSidecarsPerMinute;
+  private final int maxRecentlySampledBlocks;
   private final OptionalInt forwardSyncMaxDistanceFromHead;
 
   private SyncConfig(
@@ -55,6 +61,7 @@ private SyncConfig(
       final int forwardSyncMaxPendingBatches,
       final int forwardSyncMaxBlocksPerMinute,
       final int forwardSyncMaxBlobSidecarsPerMinute,
+      final int maxRecentlySampledBlocks,
       final OptionalInt forwardSyncMaxDistanceFromHead) {
     this.isEnabled = isEnabled;
     this.isMultiPeerSyncEnabled = isMultiPeerSyncEnabled;
@@ -65,6 +72,7 @@ private SyncConfig(
     this.forwardSyncMaxPendingBatches = forwardSyncMaxPendingBatches;
     this.forwardSyncMaxBlocksPerMinute = forwardSyncMaxBlocksPerMinute;
     this.forwardSyncMaxBlobSidecarsPerMinute = forwardSyncMaxBlobSidecarsPerMinute;
+    this.maxRecentlySampledBlocks = maxRecentlySampledBlocks;
     this.forwardSyncMaxDistanceFromHead = forwardSyncMaxDistanceFromHead;
   }
 
@@ -108,6 +116,10 @@ public int getForwardSyncMaxBlobSidecarsPerMinute() {
     return forwardSyncMaxBlobSidecarsPerMinute;
   }
 
+  public int getMaxRecentlySampledBlocks() {
+    return maxRecentlySampledBlocks;
+  }
+
   public OptionalInt getForwardSyncMaxDistanceFromHead() {
     return forwardSyncMaxDistanceFromHead;
   }
@@ -123,6 +135,7 @@ public static class Builder {
     private Integer forwardSyncMaxBlocksPerMinute = DEFAULT_FORWARD_SYNC_MAX_BLOCKS_PER_MINUTE;
     private Integer forwardSyncMaxBlobSidecarsPerMinute =
         DEFAULT_FORWARD_SYNC_MAX_BLOB_SIDECARS_PER_MINUTE;
+    private Integer maxRecentlySampledBlocks = DEFAULT_MAX_RECENTLY_SAMPLED_BLOCKS;
     private OptionalInt forwardSyncMaxDistanceFromHead = OptionalInt.empty();
 
     private Builder() {}
@@ -139,6 +152,7 @@ public SyncConfig build() {
           forwardSyncMaxPendingBatches,
           forwardSyncMaxBlocksPerMinute,
           forwardSyncMaxBlobSidecarsPerMinute,
+          maxRecentlySampledBlocks,
           forwardSyncMaxDistanceFromHead);
     }
 
@@ -214,6 +228,12 @@ public Builder reconstructHistoricStatesEnabled(
       return this;
     }
 
+    public Builder maxRecentlySampledBlocks(final Integer maxRecentlySampledBlocks) {
+      checkNotNull(maxRecentlySampledBlocks);
+      this.maxRecentlySampledBlocks = maxRecentlySampledBlocks;
+      return this;
+    }
+
     public Builder fetchAllHistoricBlocks(final boolean fetchAllHistoricBlocks) {
       this.fetchAllHistoricBlocks = fetchAllHistoricBlocks;
       return this;

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/gossip/blocks/RecentBlocksFetchService.java

@@ -26,6 +26,7 @@
 import tech.pegasys.teku.spec.datastructures.attestation.ValidatableAttestation;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool;
+import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
 import tech.pegasys.teku.statetransition.util.PendingPool;
 
 public class RecentBlocksFetchService
@@ -42,12 +43,14 @@ public class RecentBlocksFetchService
   private final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool;
   private final FetchTaskFactory fetchTaskFactory;
   private final Subscribers<BlockSubscriber> blockSubscribers = Subscribers.create(true);
+  private final DasSamplerBasic dasBasicSampler;
 
   RecentBlocksFetchService(
       final AsyncRunner asyncRunner,
       final PendingPool<SignedBeaconBlock> pendingBlockPool,
       final PendingPool<ValidatableAttestation> pendingAttestationsPool,
       final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool,
+      final DasSamplerBasic dasBasicSampler,
       final ForwardSync forwardSync,
       final FetchTaskFactory fetchTaskFactory,
       final int maxConcurrentRequests) {
@@ -56,6 +59,7 @@ public class RecentBlocksFetchService
     this.pendingBlockPool = pendingBlockPool;
     this.pendingAttestationsPool = pendingAttestationsPool;
     this.blockBlobSidecarsTrackersPool = blockBlobSidecarsTrackersPool;
+    this.dasBasicSampler = dasBasicSampler;
     this.fetchTaskFactory = fetchTaskFactory;
   }
 
@@ -64,13 +68,15 @@ public static RecentBlocksFetchService create(
       final PendingPool<SignedBeaconBlock> pendingBlocksPool,
       final PendingPool<ValidatableAttestation> pendingAttestations,
       final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool,
+      final DasSamplerBasic dasBasicSampler,
       final ForwardSync forwardSync,
       final FetchTaskFactory fetchTaskFactory) {
     return new RecentBlocksFetchService(
         asyncRunner,
         pendingBlocksPool,
         pendingAttestations,
         blockBlobSidecarsTrackersPool,
+        dasBasicSampler,
         forwardSync,
         fetchTaskFactory,
         MAX_CONCURRENT_REQUESTS);
@@ -106,6 +112,10 @@ public void requestRecentBlock(final Bytes32 blockRoot) {
       // We already have this block, waiting for blobs
       return;
     }
+    if (dasBasicSampler.containsBlock(blockRoot)) {
+      // We already have this block in DAS sampler
+      return;
+    }
     final FetchBlockTask task = createTask(blockRoot);
     if (allTasks.putIfAbsent(blockRoot, task) != null) {
       // We're already tracking this task

beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/gossip/blocks/RecentBlocksFetchServiceTest.java

@@ -43,6 +43,7 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.spec.util.DataStructureUtil;
 import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool;
+import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
 import tech.pegasys.teku.statetransition.util.PendingPool;
 
 public class RecentBlocksFetchServiceTest {
@@ -60,6 +61,8 @@ public class RecentBlocksFetchServiceTest {
   private final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool =
       mock(BlockBlobSidecarsTrackersPool.class);
 
+  private final DasSamplerBasic dasSamplerBasic = mock(DasSamplerBasic.class);
+
   private final FetchTaskFactory fetchTaskFactory = mock(FetchTaskFactory.class);
 
   private final ForwardSync forwardSync = mock(ForwardSync.class);
@@ -81,6 +84,7 @@ public void setup() {
             pendingBlockPool,
             pendingAttestationsPool,
             blockBlobSidecarsTrackersPool,
+            dasSamplerBasic,
             forwardSync,
             fetchTaskFactory,
             maxConcurrentRequests);

beacon/sync/src/testFixtures/java/tech/pegasys/teku/beacon/sync/SyncingNodeManager.java

@@ -64,6 +64,7 @@
 import tech.pegasys.teku.statetransition.block.BlockImporter;
 import tech.pegasys.teku.statetransition.block.BlockManager;
 import tech.pegasys.teku.statetransition.block.ReceivedBlockEventsChannel;
+import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
 import tech.pegasys.teku.statetransition.datacolumns.DataAvailabilitySampler;
 import tech.pegasys.teku.statetransition.execution.ExecutionPayloadManager;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoice;
@@ -224,6 +225,7 @@ public static SyncingNodeManager create(
             pendingBlocks,
             pendingAttestations,
             BlockBlobSidecarsTrackersPool.NOOP,
+            DasSamplerBasic.NOOP,
             syncService,
             fetchBlockTaskFactory);
     recentBlocksFetcher.subscribeBlockFetched(blockManager::importBlock);

eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/phase0/forkchoice/ForkChoiceTestExecutor.java

@@ -79,7 +79,7 @@
 import tech.pegasys.teku.spec.logic.common.util.AsyncBLSSignatureVerifier;
 import tech.pegasys.teku.statetransition.datacolumns.CurrentSlotProvider;
 import tech.pegasys.teku.statetransition.datacolumns.DasCustodyStand;
-import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
+import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasicImpl;
 import tech.pegasys.teku.statetransition.datacolumns.DataColumnSidecarRecoveringCustody;
 import tech.pegasys.teku.statetransition.datacolumns.retriever.DataColumnSidecarRetrieverStub;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoice;
@@ -161,8 +161,8 @@ spec, new SignedBlockAndState(anchorBlock, anchorState)),
     final StubBlobSidecarManager blobSidecarManager = new StubBlobSidecarManager(kzg);
     final CurrentSlotProvider currentSlotProvider =
         CurrentSlotProvider.create(spec, recentChainData.getStore());
-    final DasSamplerBasic dasSampler =
-        new DasSamplerBasic(
+    final DasSamplerBasicImpl dasSampler =
+        new DasSamplerBasicImpl(
             spec,
             asyncRunnerFactory.create("das", 1),
             currentSlotProvider,
@@ -173,7 +173,9 @@ spec, new SignedBlockAndState(anchorBlock, anchorState)),
             // and fetching from the config would break when not in fulu
             DasCustodyStand.createCustodyGroupCountManager(4, 8),
             recentChainData,
-            false);
+            false,
+            new StubMetricsSystem(),
+            64);
     final StubDataColumnSidecarManager dataColumnSidecarManager =
         new StubDataColumnSidecarManager(spec, recentChainData, dasSampler);
     // forkChoiceLateBlockReorgEnabled is true here always because this is the reference test