Make DataUnavailablePool always accept new blocks (Consensys#8129)

Author: tbenr

Diff for: ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/DataUnavailableBlockPool.java

@@ -38,13 +38,14 @@
  * with UNKNOWN_BLOCK and will be tracked in the block pendingPool.
  */
 public class DataUnavailableBlockPool implements FinalizedCheckpointChannel {
+  public static final int MAX_CAPACITY = 10;
   private static final Logger LOG = LogManager.getLogger();
 
   private static final Duration WAIT_BEFORE_RETRY = Duration.ofSeconds(1);
 
   // this is a queue of chain tips
   private final Queue<SignedBeaconBlock> awaitingDataAvailabilityQueue =
-      new ArrayBlockingQueue<>(10);
+      new ArrayBlockingQueue<>(MAX_CAPACITY);
   private final Spec spec;
   private final BlockManager blockManager;
   private final BlockBlobSidecarsTrackersPool blockBlobSidecarsTrackersPool;
@@ -72,9 +73,11 @@ public synchronized void addDataUnavailableBlock(final SignedBeaconBlock block)
 
     boolean wasEmpty = awaitingDataAvailabilityQueue.isEmpty();
     if (!awaitingDataAvailabilityQueue.offer(block)) {
+      final SignedBeaconBlock oldestBlock = awaitingDataAvailabilityQueue.poll();
+      awaitingDataAvailabilityQueue.add(block);
       LOG.info(
           "Discarding block {} as data unavailable retry pool capacity exceeded",
-          block::toLogString);
+          oldestBlock::toLogString);
       return;
     }
     if (wasEmpty) {

Diff for: ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/blobs/DataUnavailableBlockPoolTest.java

@@ -21,8 +21,11 @@
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 import static tech.pegasys.teku.spec.datastructures.validator.BroadcastValidationLevel.NOT_REQUIRED;
+import static tech.pegasys.teku.statetransition.blobs.DataUnavailableBlockPool.MAX_CAPACITY;
 
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.IntStream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
@@ -197,4 +200,22 @@ void shouldPruneBlocksOlderThanFinalizedSlot() {
     assertThat(dataUnavailableBlockPool.containsBlock(blockAtSlot10)).isFalse();
     assertThat(dataUnavailableBlockPool.containsBlock(blockAtSlot11)).isFalse();
   }
+
+  @Test
+  void shouldDiscardOldestWhenFull() {
+    final List<SignedBeaconBlock> blocks =
+        IntStream.range(0, MAX_CAPACITY)
+            .mapToObj(dataStructureUtil::randomSignedBeaconBlock)
+            .toList();
+    blocks.forEach(dataUnavailableBlockPool::addDataUnavailableBlock);
+
+    assertThat(dataUnavailableBlockPool.containsBlock(blocks.get(0))).isTrue();
+
+    final SignedBeaconBlock newBlock = dataStructureUtil.randomSignedBeaconBlock();
+
+    dataUnavailableBlockPool.addDataUnavailableBlock(newBlock);
+
+    assertThat(dataUnavailableBlockPool.containsBlock(blocks.get(0))).isFalse();
+    assertThat(dataUnavailableBlockPool.containsBlock(newBlock)).isTrue();
+  }
 }