Be more conservative in RPC peer rate limiting (#9841)

tbenr · web-flow · commit 5997d1fa006d · 2025-09-22T23:32:13.000+02:00
* tmp

* fix

* refactor

* rate limit via storageLimitHit

* update default limits

* fix

* fix and add tests

* tmp

* improvement

* add logs

* Revert "add logs"

This reverts commit 2b2efc740b0134e0e4b68d22fe0f22e575a1c33a.

* Revert "improvement"

This reverts commit 17206d02fc77f4b981814998048eee4b9196928f.

* Revert "tmp"

This reverts commit 5f53cf433840c5a3ad5ca102e0aa94f182f65e3c.

* revert some complexity

* revert more some complexity

* revert more some complexity

* do not adjust on error everywhere

* add comment
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/P2PConfig.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/P2PConfig.java
@@ -55,9 +55,9 @@ public class P2PConfig {
   // DEFAULT_BACKGROUND_THREAD_COUNT)
   // The storage query channel allows up to 10 parallel queries (STORAGE_QUERY_CHANNEL_PARALLELISM)
   // To avoid resource saturation and ensure capacity for other tasks, we limit historical data
-  // queries to 5
-  public static final int DEFAULT_HISTORICAL_DATA_MAX_CONCURRENT_QUERIES = 5;
-  public static final int DEFAULT_HISTORICAL_MAX_QUERY_QUEUE_SIZE = 100_000;
+  // queries to 3
+  public static final int DEFAULT_HISTORICAL_DATA_MAX_CONCURRENT_QUERIES = 3;
+  public static final int DEFAULT_HISTORICAL_MAX_QUERY_QUEUE_SIZE = 500;
 
   private final Spec spec;
   private final NetworkConfig networkConfig;
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRangeMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRangeMessageHandler.java
@@ -156,7 +156,6 @@ public void onIncomingMessage(
               callback.completeSuccessfully();
             },
             error -> {
-              peer.adjustBlocksRequest(maybeRequestKey.get(), 0);
               final Throwable rootCause = Throwables.getRootCause(error);
               if (rootCause instanceof RpcException) {
                 LOG.trace("Rejecting beacon blocks by range request", error); // Keep full context
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRootMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRootMessageHandler.java
@@ -137,10 +137,7 @@ public void onIncomingMessage(
           }
           callback.completeSuccessfully();
         },
-        err -> {
-          peer.adjustBlocksRequest(maybeRequestKey.get(), 0);
-          handleError(callback, err);
-        });
+        err -> handleError(callback, err));
   }
 
   private SafeFuture<Optional<SignedBeaconBlock>> retrieveBlock(final Bytes32 blockRoot) {
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRangeMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRangeMessageHandler.java
@@ -213,10 +213,7 @@ public void onIncomingMessage(
               LOG.trace("Sent {} blob sidecars to peer {}.", sentBlobSidecars, peer.getId());
               callback.completeSuccessfully();
             },
-            error -> {
-              peer.adjustBlobSidecarsRequest(maybeRequestKey.get(), 0);
-              handleProcessingRequestError(error, callback);
-            });
+            error -> handleProcessingRequestError(error, callback));
   }
 
   private int calculateRequestedCount(
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRootMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRootMessageHandler.java
@@ -145,10 +145,7 @@ public void onIncomingMessage(
           }
           callback.completeSuccessfully();
         },
-        err -> {
-          peer.adjustBlobSidecarsRequest(maybeRequestKey.get(), 0);
-          handleError(callback, err);
-        });
+        err -> handleError(callback, err));
   }
 
   private int getMaxRequestBlobSidecars() {
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRangeMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRangeMessageHandler.java
@@ -178,10 +178,7 @@ public void onIncomingMessage(
           }
           responseCallbackWithLogging.completeSuccessfully();
         },
-        error -> {
-          peer.adjustDataColumnSidecarsRequest(maybeRequestKey.get(), 0);
-          handleProcessingRequestError(error, responseCallbackWithLogging);
-        });
+        error -> handleProcessingRequestError(error, responseCallbackWithLogging));
   }
 
   private int calculateRequestedCount(final DataColumnSidecarsByRangeRequestMessage message) {
diff --git a/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRootMessageHandler.java b/networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRootMessageHandler.java
@@ -50,7 +50,6 @@
 import tech.pegasys.teku.statetransition.datacolumns.log.rpc.DasReqRespLogger;
 import tech.pegasys.teku.statetransition.datacolumns.log.rpc.LoggingPeerId;
 import tech.pegasys.teku.statetransition.datacolumns.log.rpc.ReqRespResponseLogger;
-import tech.pegasys.teku.storage.api.ThrottlingStorageQueryChannel;
 import tech.pegasys.teku.storage.client.CombinedChainDataClient;
 
 /**
@@ -193,11 +192,7 @@ public void onIncomingMessage(
               }
               responseCallbackWithLogging.completeSuccessfully();
             })
-        .finish(
-            err -> {
-              peer.adjustDataColumnSidecarsRequest(maybeRequestKey.get(), 0);
-              handleError(responseCallbackWithLogging, err);
-            });
+        .finish(err -> handleError(responseCallbackWithLogging, err));
   }
 
   private SafeFuture<Optional<DataColumnSidecar>> getNonCanonicalDataColumnSidecar(
@@ -233,7 +228,6 @@ private SafeFuture<Void> validateMinimumRequestEpoch(
             () ->
                 combinedChainDataClient
                     .getBlockByBlockRoot(identifier.blockRoot())
-                    .exceptionally(ThrottlingStorageQueryChannel::ignoreQueueIsFullException)
                     .thenApply(maybeBlock -> maybeBlock.map(SignedBeaconBlock::getSlot)))
         .thenAcceptChecked(
             maybeSlot -> {
@@ -264,8 +258,7 @@ private SafeFuture<Optional<DataColumnSidecar>> retrieveDataColumnSidecar(
               }
               // Fallback to non-canonical sidecar if the canonical one is not found
               return getNonCanonicalDataColumnSidecar(identifier);
-            })
-        .exceptionally(ThrottlingStorageQueryChannel::ignoreQueueIsFullException);
+            });
   }
 
   private void handleError(
diff --git a/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRangeMessageHandlerTest.java b/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRangeMessageHandlerTest.java
@@ -283,9 +283,8 @@ public void shouldReturnErrorWhenFirstBlockIsMissing() {
 
     // Requesting 5 blocks
     verify(peer, times(1)).approveBlocksRequest(any(), eq(Long.valueOf(count)));
-    // Sending 0 blocks (First block is missing, return error)
-    verify(peer, times(1))
-        .adjustBlocksRequest(eq(maybeRequestKey.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlocksRequest(any(), anyLong());
 
     final RpcException expectedError =
         new RpcException.ResourceUnavailableException(
@@ -309,9 +308,8 @@ public void shouldReturnErrorWhenEarliestHistoricalBlockUnknown() {
 
     // Requesting 5 blocks
     verify(peer, times(1)).approveBlocksRequest(any(), eq(Long.valueOf(count)));
-    // Sending 0 blocks (First block is missing, return error)
-    verify(peer, times(1))
-        .adjustBlocksRequest(eq(maybeRequestKey.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlocksRequest(any(), anyLong());
 
     final RpcException expectedError =
         new RpcException.ResourceUnavailableException(
diff --git a/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRootMessageHandlerTest.java b/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BeaconBlocksByRootMessageHandlerTest.java
@@ -156,9 +156,8 @@ public void onIncomingMessage_interruptedByClosedStream() {
 
     // Requesting 5 blocks
     verify(peer, times(1)).approveBlocksRequest(any(), eq(Long.valueOf(blocks.size())));
-    // Request cancelled
-    verify(peer, times(1))
-        .adjustBlocksRequest(eq(maybeRequestKey.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlocksRequest(any(), anyLong());
 
     // Check that we only asked for the first block
     verify(recentChainData, times(1)).retrieveSignedBlockByRoot(any());
diff --git a/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRangeMessageHandlerTest.java b/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRangeMessageHandlerTest.java
@@ -283,9 +283,8 @@ public void shouldSendResourceUnavailableIfBlobSidecarsAreNotAvailable() {
 
     // Requesting 5 * maxBlobsPerBlock blob sidecars
     verify(peer).approveBlobSidecarsRequest(any(), eq(count.times(maxBlobsPerBlock).longValue()));
-    // Request cancelled
-    verify(peer)
-        .adjustBlobSidecarsRequest(eq(allowedObjectsRequest.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlobSidecarsRequest(any(), anyLong());
 
     // blob sidecars should be available from epoch 5000, but they are
     // available from epoch 5010
diff --git a/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRootMessageHandlerTest.java b/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/BlobSidecarsByRootMessageHandlerTest.java
@@ -268,9 +268,8 @@ public void shouldSendAvailableOnlyResources() {
 
     // Requesting 3 blob sidecars
     verify(peer, times(1)).approveBlobSidecarsRequest(any(), eq(Long.valueOf(3)));
-    // Request cancelled due to error
-    verify(peer, times(1))
-        .adjustBlobSidecarsRequest(eq(allowedObjectsRequest.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlobSidecarsRequest(any(), anyLong());
 
     verify(callback, never()).respond(any());
     verify(callback).completeWithErrorResponse(rpcExceptionCaptor.capture());
@@ -305,9 +304,8 @@ public void shouldSendAvailableOnlyResources() {
 
     // Requesting 3 blob sidecars
     verify(peer, times(1)).approveBlobSidecarsRequest(any(), eq(Long.valueOf(3)));
-    // Request cancelled due to error
-    verify(peer, times(1))
-        .adjustBlobSidecarsRequest(eq(allowedObjectsRequest.orElseThrow()), eq(Long.valueOf(0)));
+    // Be protective: do not adjust due to error
+    verify(peer, never()).adjustBlobSidecarsRequest(any(), anyLong());
 
     verify(callback, never()).respond(any());
     verify(callback).completeWithErrorResponse(rpcExceptionCaptor.capture());
diff --git a/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRootMessageHandlerTest.java b/networking/eth2/src/test/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/DataColumnSidecarsByRootMessageHandlerTest.java
@@ -16,6 +16,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -39,7 +40,6 @@
 import org.junit.jupiter.api.TestTemplate;
 import org.mockito.ArgumentCaptor;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
-import tech.pegasys.teku.infrastructure.async.ThrottlingTaskQueue.QueueIsFullException;
 import tech.pegasys.teku.infrastructure.metrics.StubMetricsSystem;
 import tech.pegasys.teku.infrastructure.metrics.TekuMetricCategory;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
@@ -308,8 +308,7 @@ public void shouldSendAvailableOnlyResources() {
     // Requesting 4 data column sidecars
     verify(peer).approveDataColumnSidecarsRequest(any(), eq(Long.valueOf(4)));
     // Request cancelled due to error
-    verify(peer, times(1))
-        .adjustDataColumnSidecarsRequest(eq(allowedRequest.get()), eq(Long.valueOf(0)));
+    verify(peer, never()).adjustDataColumnSidecarsRequest(any(), anyLong());
 
     verify(callback, never()).respond(any());
     verify(callback).completeWithErrorResponse(rpcExceptionCaptor.capture());
@@ -323,64 +322,6 @@ public void shouldSendAvailableOnlyResources() {
             dataColumnsByRootIdentifiers[0].getBlockRoot());
   }
 
-  @TestTemplate
-  public void shouldIgnoreQueueIsFullExceptionFromThrottlingStorageQueryChannel() {
-    final DataColumnsByRootIdentifier[] dataColumnsByRootIdentifiers =
-        generateDataColumnsByRootIdentifiers(4, 1);
-    final List<DataColumnSidecar> generatedSidecars =
-        IntStream.range(0, 4).mapToObj(__ -> dataStructureUtil.randomDataColumnSidecar()).toList();
-
-    // simulating that the 2nd and 3rd sidecars are rejected by the throttling queue
-    final List<DataColumnSidecar> queueRejectedSidecars = generatedSidecars.subList(2, 4);
-
-    // the second block is rejected by the throttling queue
-    final Bytes32 secondBlockRoot = dataColumnsByRootIdentifiers[1].getBlockRoot();
-    when(combinedChainDataClient.getBlockByBlockRoot(secondBlockRoot))
-        .thenReturn(SafeFuture.failedFuture(new QueueIsFullException()));
-
-    when(custody.getCustodyDataColumnSidecarByRoot(any()))
-        .thenAnswer(
-            invocation -> {
-              final DataColumnIdentifier dataColumnIdentifier = invocation.getArgument(0);
-              if (dataColumnIdentifier.blockRoot().equals(secondBlockRoot)) {
-                return SafeFuture.completedFuture(Optional.empty());
-              }
-              for (int i = 0; i < 4; ++i) {
-                if (dataColumnsByRootIdentifiers[i]
-                    .getBlockRoot()
-                    .equals(dataColumnIdentifier.blockRoot())) {
-                  if (queueRejectedSidecars.contains(generatedSidecars.get(i))) {
-                    return SafeFuture.failedFuture(new QueueIsFullException());
-                  }
-                  return SafeFuture.completedFuture(Optional.of(generatedSidecars.get(i)));
-                }
-              }
-              throw new RuntimeException("Should never get here");
-            });
-
-    handler.onIncomingMessage(
-        protocolId, peer, messageSchema.of(dataColumnsByRootIdentifiers), callback);
-
-    // Requesting 4 data column sidecars
-    verify(peer).approveDataColumnSidecarsRequest(any(), eq(Long.valueOf(4)));
-    // Sending 1 data column sidecars
-    verify(peer).adjustDataColumnSidecarsRequest(eq(allowedRequest.get()), eq(Long.valueOf(1)));
-
-    verify(combinedChainDataClient, never()).getNonCanonicalSidecar(any());
-    verify(callback, times(1)).respond(datacolumnSidecarCaptor.capture());
-    verify(callback).completeSuccessfully();
-
-    final List<Bytes32> respondedDataColumnSidecarBlockRoots =
-        datacolumnSidecarCaptor.getAllValues().stream()
-            .map(DataColumnSidecar::getBlockRoot)
-            .toList();
-    final List<Bytes32> expectedDataColumnIdentifiersBlockRoots =
-        List.of(generatedSidecars.get(0).getBlockRoot());
-
-    assertThat(respondedDataColumnSidecarBlockRoots)
-        .containsExactlyElementsOf(expectedDataColumnIdentifiersBlockRoots);
-  }
-
   @TestTemplate
   public void shouldSendCustodyOnlyResources() {
     final DataColumnsByRootIdentifier[] dataColumnsByRootIdentifiers =
@@ -492,6 +433,25 @@ public void shouldSendToPeerRequestedBlobSidecars() {
         .containsExactlyElementsOf(expectedDataColumnIdentifiersBlockRoots);
   }
 
+  @TestTemplate
+  public void shouldNotAdjustIfAnErrorOccurs() {
+    // Be protective: do not adjust due to error
+    final DataColumnsByRootIdentifier[] dataColumnsByRootIdentifiers =
+        generateDataColumnsByRootIdentifiers(4, 1);
+
+    final RuntimeException error = new RuntimeException("Fatal error");
+
+    when(custody.getCustodyDataColumnSidecarByRoot(any()))
+        .thenReturn(SafeFuture.failedFuture(error));
+
+    handler.onIncomingMessage(
+        protocolId, peer, messageSchema.of(dataColumnsByRootIdentifiers), callback);
+
+    verify(callback)
+        .completeWithUnexpectedError(argThat(exception -> exception.getCause().equals(error)));
+    verify(peer, never()).adjustDataColumnSidecarsRequest(any(), anyLong());
+  }
+
   private DataColumnsByRootIdentifier[] generateDataColumnsByRootIdentifiers(
       final int numberOfBlocks, final int columnIndicesPerBlock) {
     return IntStream.range(0, numberOfBlocks)
diff --git a/storage/api/src/main/java/tech/pegasys/teku/storage/api/ThrottlingStorageQueryChannel.java b/storage/api/src/main/java/tech/pegasys/teku/storage/api/ThrottlingStorageQueryChannel.java
@@ -13,13 +13,10 @@
 
 package tech.pegasys.teku.storage.api;
 
-import static tech.pegasys.teku.infrastructure.async.ThrottlingTaskQueue.isQueueIsFullException;
-
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.CompletionException;
 import org.apache.tuweni.bytes.Bytes32;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 import tech.pegasys.teku.ethereum.pow.api.DepositTreeSnapshot;
@@ -60,16 +57,6 @@ public ThrottlingStorageQueryChannel(
             "throttling_storage_query_rejected");
   }
 
-  public static <T> Optional<T> ignoreQueueIsFullException(final Throwable error) {
-    if (isQueueIsFullException(error)) {
-      return Optional.empty();
-    }
-    if (error instanceof RuntimeException exception) {
-      throw exception;
-    }
-    throw new CompletionException(error);
-  }
-
   @Override
   public SafeFuture<Optional<OnDiskStoreData>> onStoreRequest() {
     return taskQueue.queueTask(delegate::onStoreRequest);

Original file line number	Diff line number	Diff line change
`@@ -137,10 +137,7 @@ public void onIncomingMessage(`
`137`	`137`	`}`
`138`	`138`	`callback.completeSuccessfully();`
`139`	`139`	`},`
`140`		`- err -> {`
`141`		`- peer.adjustBlocksRequest(maybeRequestKey.get(), 0);`
`142`		`- handleError(callback, err);`
`143`		`- });`
	`140`	`+ err -> handleError(callback, err));`
`144`	`141`	`}`
`145`	`142`
`146`	`143`	`private SafeFuture<Optional<SignedBeaconBlock>> retrieveBlock(final Bytes32 blockRoot) {`
Original file line number	Diff line number	Diff line change
`@@ -145,10 +145,7 @@ public void onIncomingMessage(`
`145`	`145`	`}`
`146`	`146`	`callback.completeSuccessfully();`
`147`	`147`	`},`
`148`		`- err -> {`
`149`		`- peer.adjustBlobSidecarsRequest(maybeRequestKey.get(), 0);`
`150`		`- handleError(callback, err);`
`151`		`- });`
	`148`	`+ err -> handleError(callback, err));`
`152`	`149`	`}`
`153`	`150`
`154`	`151`	`private int getMaxRequestBlobSidecars() {`
Original file line number	Diff line number	Diff line change
`@@ -178,10 +178,7 @@ public void onIncomingMessage(`
`178`	`178`	`}`
`179`	`179`	`responseCallbackWithLogging.completeSuccessfully();`
`180`	`180`	`},`
`181`		`- error -> {`
`182`		`- peer.adjustDataColumnSidecarsRequest(maybeRequestKey.get(), 0);`
`183`		`- handleProcessingRequestError(error, responseCallbackWithLogging);`
`184`		`- });`
	`181`	`+ error -> handleProcessingRequestError(error, responseCallbackWithLogging));`
`185`	`182`	`}`
`186`	`183`
`187`	`184`	`private int calculateRequestedCount(final DataColumnSidecarsByRangeRequestMessage message) {`