tweaks (#10961)

Author: rolfyone

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/validation/AttestationValidator.java

@@ -87,7 +87,10 @@ public SafeFuture<InternalValidationResult> validateSingleOrAggregateAttestation
       return SafeFuture.completedFuture(InternalValidationResult.ACCEPT);
     }
     return singleOrAggregateAttestationChecks(
-            signatureVerifier, validatableAttestation, validatableAttestation.getReceivedSubnetId())
+            signatureVerifier,
+            validatableAttestation,
+            validatableAttestation.getReceivedSubnetId(),
+            true)
         .thenApply(InternalValidationResultWithState::getResult)
         .thenPeek(
             result -> {
@@ -116,6 +119,19 @@ SafeFuture<InternalValidationResultWithState> singleOrAggregateAttestationChecks
       final AsyncBLSSignatureVerifier signatureVerifier,
       final ValidatableAttestation validatableAttestation,
       final OptionalInt receivedOnSubnetId) {
+    // Aggregate validation path: preserve legacy behaviour where a future-slot attestation is
+    // deferred without running signature verification here. The aggregate wrapper uses a batch
+    // verifier that is flushed separately, so verifying (and optimistically caching) the signature
+    // here would leave an unverified signature cached if the deferral short-circuits the flush.
+    return singleOrAggregateAttestationChecks(
+        signatureVerifier, validatableAttestation, receivedOnSubnetId, false);
+  }
+
+  SafeFuture<InternalValidationResultWithState> singleOrAggregateAttestationChecks(
+      final AsyncBLSSignatureVerifier signatureVerifier,
+      final ValidatableAttestation validatableAttestation,
+      final OptionalInt receivedOnSubnetId,
+      final boolean verifyFutureSlotAttestationSignature) {
 
     Attestation attestation = validatableAttestation.getAttestation();
     final AttestationData data = attestation.getData();
@@ -165,10 +181,20 @@ SafeFuture<InternalValidationResultWithState> singleOrAggregateAttestationChecks
             gossipValidationHelper.getCurrentTimeMillis());
 
     if (slotInclusionGossipValidationResult.isPresent()) {
-      return switch (slotInclusionGossipValidationResult.get()) {
-        case IGNORE -> completedFuture(InternalValidationResultWithState.ignore());
-        case SAVE_FOR_FUTURE -> completedFuture(InternalValidationResultWithState.saveForFuture());
-      };
+      if (slotInclusionGossipValidationResult.get() == SlotInclusionGossipValidationResult.IGNORE) {
+        return completedFuture(InternalValidationResultWithState.ignore());
+      }
+      // SAVE_FOR_FUTURE: a future-slot attestation is deferred rather than accepted for gossip.
+      // When requested, we still verify its signature so that a bogus signature is rejected and its
+      // sender penalised, rather than being deferred with a penalty-free Ignore verdict and later
+      // forcing a BLS verification in the fork choice path for free. Only an invalid signature
+      // rejects here; every other gossip check simply defers (fork choice re-validates the rest
+      // once the attestation's slot arrives). A verified signature is cached so fork choice does
+      // not re-verify it.
+      if (!verifyFutureSlotAttestationSignature) {
+        return completedFuture(InternalValidationResultWithState.saveForFuture());
+      }
+      return checkFutureSlotAttestationSignature(validatableAttestation, signatureVerifier);
     }
 
     // [REJECT] The block being voted for (attestation.data.beacon_block_root) passes validation.
@@ -276,4 +302,40 @@ SafeFuture<InternalValidationResultWithState> singleOrAggregateAttestationChecks
                       });
             });
   }
+
+  /**
+   * Verifies the signature of a future-slot attestation and always defers it (SAVE_FOR_FUTURE),
+   * rejecting only when the signature is invalid. The remaining gossip checks are intentionally
+   * skipped here: they either only apply to attestations eligible for immediate propagation, or are
+   * re-validated by fork choice when the attestation's slot arrives. Rejecting solely on an invalid
+   * signature keeps the peer penalty limited to provably-invalid messages, matching how other
+   * clients score future-slot attestations, while still avoiding a penalty-free BLS verification in
+   * the fork choice path.
+   */
+  private SafeFuture<InternalValidationResultWithState> checkFutureSlotAttestationSignature(
+      final ValidatableAttestation validatableAttestation,
+      final AsyncBLSSignatureVerifier signatureVerifier) {
+    final AttestationData data = validatableAttestation.getAttestation().getData();
+    // The signature can't be verified without the block being voted for and its state; defer.
+    if (!gossipValidationHelper.isBlockAvailable(data.getBeaconBlockRoot())) {
+      return completedFuture(InternalValidationResultWithState.saveForFuture());
+    }
+    return gossipValidationHelper
+        .getStateForAttestationValidation(data)
+        .thenCompose(
+            maybeState -> {
+              if (maybeState.isEmpty()) {
+                return completedFuture(InternalValidationResultWithState.saveForFuture());
+              }
+              return spec.isValidIndexedAttestation(
+                      maybeState.get(), validatableAttestation, signatureVerifier)
+                  .thenApply(
+                      signatureResult ->
+                          signatureResult.isSuccessful()
+                              ? InternalValidationResultWithState.saveForFuture()
+                              : InternalValidationResultWithState.reject(
+                                  "Attestation is not a valid indexed attestation: %s",
+                                  signatureResult.getInvalidReason()));
+            });
+  }
 }

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/validation/Phase0AttestationValidatorTest.java

@@ -19,6 +19,7 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static tech.pegasys.teku.infrastructure.unsigned.UInt64.ONE;
 import static tech.pegasys.teku.infrastructure.unsigned.UInt64.ZERO;
@@ -364,4 +365,72 @@ public void shouldRejectAttestationsThatHaveLMDVotesInconsistentWithTargetRoot()
         .matches(
             rejected("descend from target block"), "Rejected does not descend from target block");
   }
+
+  @Test
+  public void shouldRejectFutureSlotAttestationWithInvalidSignature() {
+    // A future-slot attestation with a bogus signature must be rejected (so the sender is
+    // penalised) rather than deferred with a penalty-free Ignore verdict, which would force a
+    // signature verification in the fork choice path for free.
+    final StateAndBlockSummary blockAndState = storageSystem.getChainHead();
+    final Attestation attestation = attestationGenerator.validAttestation(blockAndState, ONE);
+    assertThat(attestation.getData().getSlot()).isEqualTo(ONE);
+
+    final AsyncBLSSignatureVerifier signatureVerifier = mock(AsyncBLSSignatureVerifier.class);
+    when(signatureVerifier.verify(anyList(), any(Bytes.class), any()))
+        .thenReturn(SafeFuture.completedFuture(false));
+    final AttestationValidator validator =
+        new AttestationValidator(
+            spec, signatureVerifier, gossipValidationHelper, invalidBlockRoots);
+
+    final int subnetId = spec.computeSubnetForAttestation(blockAndState.getState(), attestation);
+    chainUpdater.setCurrentSlot(ZERO);
+
+    assertThat(
+            validator
+                .validate(ValidatableAttestation.fromNetwork(spec, attestation, subnetId))
+                .join()
+                .code())
+        .isEqualTo(REJECT);
+    // The signature was actually verified at gossip time, giving a REJECT verdict that penalises
+    // the sender.
+    verify(signatureVerifier).verify(anyList(), any(Bytes.class), any());
+  }
+
+  @Test
+  public void shouldVerifyAndCacheSignatureForValidFutureSlotAttestation() {
+    // A valid future-slot attestation is still only deferred (SAVE_FOR_FUTURE), but its signature
+    // is verified and cached here so the fork choice path does not re-verify it.
+    final StateAndBlockSummary blockAndState = storageSystem.getChainHead();
+    final Attestation attestation = attestationGenerator.validAttestation(blockAndState, ONE);
+    assertThat(attestation.getData().getSlot()).isEqualTo(ONE);
+    chainUpdater.setCurrentSlot(ZERO);
+
+    final ValidatableAttestation validatableAttestation =
+        ValidatableAttestation.fromNetwork(
+            spec,
+            attestation,
+            spec.computeSubnetForAttestation(blockAndState.getState(), attestation));
+
+    assertThat(validator.validate(validatableAttestation).join().code()).isEqualTo(SAVE_FOR_FUTURE);
+    assertThat(validatableAttestation.isValidIndexedAttestation()).isTrue();
+  }
+
+  @Test
+  public void shouldDeferFutureSlotAttestationOnWrongSubnetWhenSignatureIsValid() {
+    // A future-slot attestation is only rejected for an invalid signature; other gossip failures
+    // (here, the wrong subnet) defer rather than penalise the sender. Contrast with
+    // shouldRejectAttestationsSentOnTheWrongSubnet, which rejects an in-window attestation.
+    final StateAndBlockSummary blockAndState = storageSystem.getChainHead();
+    final Attestation attestation = attestationGenerator.validAttestation(blockAndState, ONE);
+    assertThat(attestation.getData().getSlot()).isEqualTo(ONE);
+    final int subnetId = spec.computeSubnetForAttestation(blockAndState.getState(), attestation);
+    chainUpdater.setCurrentSlot(ZERO);
+
+    assertThat(
+            validator
+                .validate(ValidatableAttestation.fromNetwork(spec, attestation, subnetId + 1))
+                .join()
+                .code())
+        .isEqualTo(SAVE_FOR_FUTURE);
+  }
 }

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/LibP2PNetwork.java

@@ -48,6 +48,7 @@ public class LibP2PNetwork implements P2PNetwork<Peer> {
   private static final Logger LOG = LogManager.getLogger();
   static final int REMOTE_OPEN_STREAMS_RATE_LIMIT = 256;
   static final int REMOTE_PARALLEL_OPEN_STREAMS_COUNT_LIMIT = 256;
+  static final int MAXIMUM_ACTIVE_INBOUND_RPC_STREAMS = 128;
 
   private final PrivKey privKey;
   private final NodeId nodeId;

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/LibP2PNetworkBuilder.java

@@ -13,6 +13,7 @@
 
 package tech.pegasys.teku.networking.p2p.libp2p;
 
+import static tech.pegasys.teku.networking.p2p.libp2p.LibP2PNetwork.MAXIMUM_ACTIVE_INBOUND_RPC_STREAMS;
 import static tech.pegasys.teku.networking.p2p.libp2p.LibP2PNetwork.REMOTE_OPEN_STREAMS_RATE_LIMIT;
 import static tech.pegasys.teku.networking.p2p.libp2p.LibP2PNetwork.REMOTE_PARALLEL_OPEN_STREAMS_COUNT_LIMIT;
 
@@ -49,6 +50,7 @@
 import tech.pegasys.teku.networking.p2p.libp2p.gossip.GossipTopicFilter;
 import tech.pegasys.teku.networking.p2p.libp2p.gossip.LibP2PGossipNetwork;
 import tech.pegasys.teku.networking.p2p.libp2p.gossip.LibP2PGossipNetworkBuilder;
+import tech.pegasys.teku.networking.p2p.libp2p.rpc.InboundRpcStreamLimiter;
 import tech.pegasys.teku.networking.p2p.libp2p.rpc.RpcHandler;
 import tech.pegasys.teku.networking.p2p.network.P2PNetwork;
 import tech.pegasys.teku.networking.p2p.network.PeerHandler;
@@ -289,7 +291,11 @@ private static boolean shouldListenOnQuic(
   }
 
   protected List<? extends RpcHandler<?, ?, ?>> createRpcHandlers() {
-    return rpcMethods.stream().map(m -> new RpcHandler<>(asyncRunner, m, metricsSystem)).toList();
+    final InboundRpcStreamLimiter inboundRpcStreamLimiter =
+        new InboundRpcStreamLimiter(MAXIMUM_ACTIVE_INBOUND_RPC_STREAMS);
+    return rpcMethods.stream()
+        .map(m -> new RpcHandler<>(asyncRunner, m, metricsSystem, inboundRpcStreamLimiter))
+        .toList();
   }
 
   protected LibP2PGossipNetwork createGossipNetwork() {

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/rpc/InboundRpcStreamLimiter.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Consensys Software Inc., 2026
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package tech.pegasys.teku.networking.p2p.libp2p.rpc;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** Limits the number of active inbound RPC streams across every transport and connection. */
+public class InboundRpcStreamLimiter {
+
+  private final int maximumActiveStreams;
+  private final AtomicInteger activeStreams = new AtomicInteger();
+
+  public InboundRpcStreamLimiter(final int maximumActiveStreams) {
+    if (maximumActiveStreams < 1) {
+      throw new IllegalArgumentException("Maximum active inbound RPC streams must be positive");
+    }
+    this.maximumActiveStreams = maximumActiveStreams;
+  }
+
+  public boolean tryAcquire() {
+    int currentActiveStreams;
+    do {
+      currentActiveStreams = activeStreams.get();
+      if (currentActiveStreams >= maximumActiveStreams) {
+        return false;
+      }
+    } while (!activeStreams.compareAndSet(currentActiveStreams, currentActiveStreams + 1));
+    return true;
+  }
+
+  public void release() {
+    final int remainingStreams = activeStreams.decrementAndGet();
+    if (remainingStreams < 0) {
+      activeStreams.incrementAndGet();
+      throw new IllegalStateException("Released an inbound RPC stream that was not acquired");
+    }
+  }
+
+  public int getMaximumActiveStreams() {
+    return maximumActiveStreams;
+  }
+
+  int getActiveStreams() {
+    return activeStreams.get();
+  }
+}

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/rpc/RpcHandler.java

@@ -70,6 +70,7 @@ public class RpcHandler<
 
   private final AsyncRunner asyncRunner;
   private final RpcMethod<TOutgoingHandler, TRequest, TRespHandler> rpcMethod;
+  private final InboundRpcStreamLimiter inboundRpcStreamLimiter;
 
   final Counter rpcRequestsTotalCounter;
   final Counter rpcRequestsFailedCounter;
@@ -78,9 +79,11 @@ public class RpcHandler<
   public RpcHandler(
       final AsyncRunner asyncRunner,
       final RpcMethod<TOutgoingHandler, TRequest, TRespHandler> rpcMethod,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
+      final InboundRpcStreamLimiter inboundRpcStreamLimiter) {
     this.asyncRunner = asyncRunner;
     this.rpcMethod = rpcMethod;
+    this.inboundRpcStreamLimiter = inboundRpcStreamLimiter;
 
     rpcRequestsTotalCounter =
         metricsSystem.createCounter(
@@ -206,13 +209,37 @@ public SafeFuture<Controller<TOutgoingHandler>> initChannel(
     final Connection connection = streamChannel.getConnection();
     final NodeId nodeId = new LibP2PNodeId(connection.secureSession().getRemoteId());
 
-    final Controller<TOutgoingHandler> controller = new Controller<>(nodeId, streamChannel);
-    if (!channel.isInitiator()) {
-      controller.setIncomingRequestHandler(
-          rpcMethod.createIncomingRequestHandler(selectedProtocol));
+    final boolean isIncomingStream = !channel.isInitiator();
+    if (isIncomingStream && !inboundRpcStreamLimiter.tryAcquire()) {
+      LOG.debug(
+          "Rejecting inbound RPC stream for protocol {} from {} because the global active stream limit of {} has been reached",
+          selectedProtocol,
+          nodeId,
+          inboundRpcStreamLimiter.getMaximumActiveStreams());
+      ignoreFuture(streamChannel.close());
+      return SafeFuture.failedFuture(
+          new IllegalStateException("Maximum active inbound RPC streams reached"));
+    }
+
+    try {
+      final Controller<TOutgoingHandler> controller =
+          new Controller<>(
+              nodeId,
+              streamChannel,
+              isIncomingStream ? inboundRpcStreamLimiter::release : () -> {});
+      if (isIncomingStream) {
+        controller.setIncomingRequestHandler(
+            rpcMethod.createIncomingRequestHandler(selectedProtocol));
+      }
+      channel.pushHandler(controller);
+      return controller.activeFuture;
+    } catch (final Throwable t) {
+      if (isIncomingStream) {
+        inboundRpcStreamLimiter.release();
+      }
+      ignoreFuture(streamChannel.close());
+      return SafeFuture.failedFuture(t);
     }
-    channel.pushHandler(controller);
-    return controller.activeFuture;
   }
 
   static class Controller<TOutgoingHandler extends RpcRequestHandler>
@@ -221,16 +248,19 @@ static class Controller<TOutgoingHandler extends RpcRequestHandler>
 
     private final NodeId nodeId;
     private final Stream p2pStream;
+    private final Runnable releaseStreamPermit;
     private Optional<TOutgoingHandler> outgoingRequestHandler = Optional.empty();
     private Optional<RpcRequestHandler> rpcRequestHandler = Optional.empty();
     private RpcStream rpcStream;
     private boolean readCompleted = false;
 
     protected final SafeFuture<Controller<TOutgoingHandler>> activeFuture = new SafeFuture<>();
 
-    private Controller(final NodeId nodeId, final Stream p2pStream) {
+    private Controller(
+        final NodeId nodeId, final Stream p2pStream, final Runnable releaseStreamPermit) {
       this.nodeId = nodeId;
       this.p2pStream = p2pStream;
+      this.releaseStreamPermit = releaseStreamPermit;
     }
 
     @Override
@@ -321,6 +351,7 @@ private void onChannelClosed() {
         runHandler(h -> h.closed(nodeId, rpcStream));
       } finally {
         rpcRequestHandler = Optional.empty();
+        releaseStreamPermit.run();
       }
     }