9039 il pool (#9062)

* implement inclusion list by committee indices rpc logic

Author: mehdi-aouadi

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/inclusionlist/InclusionListManager.java

@@ -32,7 +32,6 @@ public class InclusionListManager implements SlotEventsChannel {
 
   private final SignedInclusionListValidator signedInclusionListValidator;
 
-  // TODO EIP7805 Could use a more efficient DS
   private final NavigableMap<UInt64, Map<UInt64, List<SignedInclusionList>>>
       slotToInclusionListsByValidatorIndex = new TreeMap<>();
 
@@ -81,8 +80,11 @@ public List<SignedInclusionList> getInclusionLists(
     final Map<UInt64, List<SignedInclusionList>> inclusionListsForSlot =
         slotToInclusionListsByValidatorIndex.getOrDefault(slot, Map.of());
     return inclusionListsForSlot.entrySet().stream()
-        .filter(e -> committeeIndices.isSet(e.getKey().intValue()))
-        .flatMap(e -> e.getValue().stream())
+        .filter(
+            validatorIndexToInclusionLists ->
+                committeeIndices.isSet(validatorIndexToInclusionLists.getKey().intValue()))
+        .flatMap(
+            validatorIndexToInclusionLists -> validatorIndexToInclusionLists.getValue().stream())
         .toList();
   }
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/peers/DefaultEth2Peer.java

@@ -407,6 +407,13 @@ public void adjustBlobSidecarsRequest(
         blobSidecarsRequestTracker, blobSidecarsRequest, returnedBlobSidecarsCount);
   }
 
+  @Override
+  public void adjustInclusionListsRequest(
+      final RequestApproval inclusionListsRequest, final long returnedInclusionListsCount) {
+    adjustObjectsRequest(
+        inclusionListRequestTracker, inclusionListsRequest, returnedInclusionListsCount);
+  }
+
   @Override
   public boolean approveRequest() {
     if (requestTracker.approveObjectsRequest(1L).isEmpty()) {

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/peers/Eth2Peer.java

@@ -124,6 +124,9 @@ Optional<RequestApproval> approveInclusionListsRequest(
   void adjustBlobSidecarsRequest(
       RequestApproval blobSidecarsRequest, long returnedBlobSidecarsCount);
 
+  void adjustInclusionListsRequest(
+      RequestApproval inclusionListsRequest, long returnedInclusionListsCount);
+
   boolean approveRequest();
 
   SafeFuture<UInt64> sendPing();

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/beaconchain/methods/InclusionListByCommitteeIndicesMessageHandler.java

@@ -15,21 +15,24 @@
 
 import static tech.pegasys.teku.networking.eth2.rpc.core.RpcResponseStatus.INVALID_REQUEST_CODE;
 
+import com.google.common.base.Throwables;
+import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Optional;
-import java.util.stream.Collectors;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 import org.hyperledger.besu.plugin.services.metrics.Counter;
 import org.hyperledger.besu.plugin.services.metrics.LabelledMetric;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.metrics.TekuMetricCategory;
-import tech.pegasys.teku.infrastructure.ssz.primitive.SszBit;
 import tech.pegasys.teku.networking.eth2.peers.Eth2Peer;
 import tech.pegasys.teku.networking.eth2.peers.RequestApproval;
 import tech.pegasys.teku.networking.eth2.rpc.core.PeerRequiredLocalMessageHandler;
 import tech.pegasys.teku.networking.eth2.rpc.core.ResponseCallback;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException;
+import tech.pegasys.teku.networking.p2p.rpc.StreamClosedException;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.config.SpecConfigEip7805;
 import tech.pegasys.teku.spec.datastructures.networking.libp2p.rpc.InclusionListByCommitteeRequestMessage;
@@ -72,9 +75,10 @@ public Optional<RpcException> validateRequest(
     final SpecConfigEip7805 specConfig =
         SpecConfigEip7805.required(spec.atSlot(request.getSlot()).getConfig());
 
+    final int requestedCount = request.getCommitteeIndices().getAllSetBits().size();
     final int maxRequestInclusionList = specConfig.getMaxRequestInclusionList();
 
-    if (request.size() > maxRequestInclusionList) {
+    if (requestedCount > maxRequestInclusionList) {
       requestCounter.labels("count_too_big").inc();
       return Optional.of(
           new RpcException(
@@ -86,8 +90,6 @@ public Optional<RpcException> validateRequest(
     return Optional.empty();
   }
 
-  // TODO EIP7805 review logic, add counter
-  @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   protected void onIncomingMessage(
       final String protocolId,
@@ -96,28 +98,64 @@ protected void onIncomingMessage(
       final ResponseCallback<SignedInclusionList> callback) {
 
     LOG.trace(
-        "Peer {} requested Inclusion Lists for slot {} with committee indices {}",
+        "Peer {} requested Inclusion Lists for slot {} for committee indices {}",
         peer.getId(),
         message.getSlot(),
-        message.getCommitteeIndices().stream()
-            .map(SszBit::toString)
-            .collect(Collectors.joining(",")));
+        message.getCommitteeIndices().getAllSetBits().intStream());
 
+    final int requestedCount = message.getCommitteeIndices().getAllSetBits().size();
     final Optional<RequestApproval> inclusionListsRequestApproval =
-        peer.approveInclusionListsRequest(callback, message.size());
+        peer.approveInclusionListsRequest(callback, requestedCount);
 
     if (!peer.approveRequest() || inclusionListsRequestApproval.isEmpty()) {
       requestCounter.labels("rate_limited").inc();
       return;
     }
 
     requestCounter.labels("ok").inc();
-    totalInclusionListsRequestedCounter.inc(message.size());
+    totalInclusionListsRequestedCounter.inc(requestedCount);
 
-    // TODO EIP7805 review logic / handle errors
+    final AtomicInteger sentInclusionLists = new AtomicInteger(0);
     final List<SignedInclusionList> signedInclusionLists =
         inclusionListManager.getInclusionLists(message.getSlot(), message.getCommitteeIndices());
-    signedInclusionLists.forEach(callback::respond);
-    callback.completeSuccessfully();
+    SafeFuture<Void> future = SafeFuture.COMPLETE;
+    for (SignedInclusionList signedInclusionList : signedInclusionLists) {
+      future =
+          future.thenCompose(
+              __ ->
+                  callback
+                      .respond(signedInclusionList)
+                      .thenRun(sentInclusionLists::incrementAndGet));
+    }
+    future.finish(
+        () -> {
+          if (sentInclusionLists.get() != requestedCount) {
+            peer.adjustInclusionListsRequest(
+                inclusionListsRequestApproval.get(), sentInclusionLists.get());
+          }
+          callback.completeSuccessfully();
+        },
+        err -> {
+          peer.adjustInclusionListsRequest(inclusionListsRequestApproval.get(), 0);
+          handleError(callback, err);
+        });
+  }
+
+  private void handleError(
+      final ResponseCallback<SignedInclusionList> callback, final Throwable error) {
+    final Throwable rootCause = Throwables.getRootCause(error);
+    if (rootCause instanceof RpcException) {
+      LOG.trace(
+          "Rejecting inclusion lists by committee indices request", error); // Keep full context
+      callback.completeWithErrorResponse((RpcException) rootCause);
+    } else {
+      if (rootCause instanceof StreamClosedException
+          || rootCause instanceof ClosedChannelException) {
+        LOG.trace("Stream closed while sending requested inclusion lists", error);
+      } else {
+        LOG.error("Failed to process inclusion lists by committee indices request", error);
+      }
+      callback.completeWithUnexpectedError(error);
+    }
   }
 }

networking/eth2/src/testFixtures/java/tech/pegasys/teku/networking/eth2/peers/RespondingEth2Peer.java

@@ -349,6 +349,10 @@ public Optional<RequestApproval> approveInclusionListsRequest(
   public void adjustBlobSidecarsRequest(
       final RequestApproval blobSidecarRequests, final long returnedBlobSidecarsCount) {}
 
+  @Override
+  public void adjustInclusionListsRequest(
+      final RequestApproval inclusionListsRequest, final long returnedInclusionListsCount) {}
+
   @Override
   public boolean approveRequest() {
     return true;