add engine get blobs

ethereum/executionclient/src/integration-test/java/tech/pegasys/teku/ethereum/executionclient/web3j/Web3JExecutionEngineClientTest.java

@@ -48,6 +48,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
 import tech.pegasys.teku.ethereum.events.ExecutionClientEventsChannel;
+import tech.pegasys.teku.ethereum.executionclient.schema.BlobAndProofV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ClientVersionV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV3;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV4;
@@ -66,6 +67,7 @@
 import tech.pegasys.teku.spec.SpecMilestone;
 import tech.pegasys.teku.spec.TestSpecContext;
 import tech.pegasys.teku.spec.TestSpecInvocationContextProvider.SpecContext;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar;
 import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayload;
 import tech.pegasys.teku.spec.executionlayer.PayloadStatus;
 import tech.pegasys.teku.spec.logic.versions.deneb.types.VersionedHash;
@@ -354,6 +356,50 @@ public void newPayloadV4_shouldBuildRequestAndResponseSuccessfully() {
         .isEqualTo(parentBeaconBlockRoot.toHexString());
   }
 
+  @TestTemplate
+  @SuppressWarnings("unchecked")
+  public void getBlobsV1_shouldBuildRequestAndResponseSuccessfully() {
+    assumeThat(specMilestone).isGreaterThanOrEqualTo(DENEB);
+    final List<BlobSidecar> blobSidecars =
+        dataStructureUtil.randomBlobSidecars(spec.getMaxBlobsPerBlock().orElseThrow());
+    final List<BlobAndProofV1> blobsAndProofsV1 =
+        blobSidecars.stream()
+            .map(
+                blobSidecar ->
+                    new BlobAndProofV1(
+                        blobSidecar.getBlob().getBytes(),
+                        blobSidecar.getKZGProof().getBytesCompressed()))
+            .toList();
+    final String blobsAndProofsJson =
+        blobSidecars.stream()
+            .map(
+                blobSidecar ->
+                    String.format(
+                        "{ \"blob\": \"%s\", \"proof\": \"%s\" }",
+                        blobSidecar.getBlob().getBytes().toHexString(),
+                        blobSidecar.getKZGProof().getBytesCompressed().toHexString()))
+            .collect(Collectors.joining(", "));
+    final String bodyResponse =
+        "{\"jsonrpc\": \"2.0\", \"id\": 0, \"result\": [" + blobsAndProofsJson + "]}";
+
+    mockSuccessfulResponse(bodyResponse);
+
+    final List<VersionedHash> blobVersionedHashes = dataStructureUtil.randomVersionedHashes(3);
+
+    final SafeFuture<Response<List<BlobAndProofV1>>> futureResponse =
+        eeClient.getBlobsV1(blobVersionedHashes);
+
+    assertThat(futureResponse)
+        .succeedsWithin(1, TimeUnit.SECONDS)
+        .matches(response -> response.getPayload().equals(blobsAndProofsV1));
+
+    final Map<String, Object> requestData = takeRequest();
+    verifyJsonRpcMethodCall(requestData, "engine_getBlobsV1");
+    assertThat(requestData.get("params"))
+        .asInstanceOf(LIST)
+        .containsExactly(blobVersionedHashes.stream().map(VersionedHash::toHexString).toList());
+  }
+
   private void mockSuccessfulResponse(final String responseBody) {
     mockWebServer.enqueue(
         new MockResponse()

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/ExecutionEngineClient.java

@@ -16,6 +16,7 @@
 import java.util.List;
 import java.util.Optional;
 import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.ethereum.executionclient.schema.BlobAndProofV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ClientVersionV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV2;
@@ -77,4 +78,6 @@ SafeFuture<Response<ForkChoiceUpdatedResult>> forkChoiceUpdatedV3(
   SafeFuture<Response<List<String>>> exchangeCapabilities(List<String> capabilities);
 
   SafeFuture<Response<List<ClientVersionV1>>> getClientVersionV1(ClientVersionV1 clientVersion);
+
+  SafeFuture<Response<List<BlobAndProofV1>>> getBlobsV1(List<VersionedHash> blobVersionedHashes);
 }

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/ThrottlingExecutionEngineClient.java

@@ -17,6 +17,7 @@
 import java.util.Optional;
 import org.apache.tuweni.bytes.Bytes32;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import tech.pegasys.teku.ethereum.executionclient.schema.BlobAndProofV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ClientVersionV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV2;
@@ -150,4 +151,10 @@ public SafeFuture<Response<List<ClientVersionV1>>> getClientVersionV1(
       final ClientVersionV1 clientVersion) {
     return taskQueue.queueTask(() -> delegate.getClientVersionV1(clientVersion));
   }
+
+  @Override
+  public SafeFuture<Response<List<BlobAndProofV1>>> getBlobsV1(
+      final List<VersionedHash> blobVersionedHashes) {
+    return taskQueue.queueTask(() -> delegate.getBlobsV1(blobVersionedHashes));
+  }
 }

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/methods/EngineApiMethod.java

@@ -16,7 +16,8 @@
 public enum EngineApiMethod {
   ENGINE_NEW_PAYLOAD("engine_newPayload"),
   ENGINE_GET_PAYLOAD("engine_getPayload"),
-  ENGINE_FORK_CHOICE_UPDATED("engine_forkchoiceUpdated");
+  ENGINE_FORK_CHOICE_UPDATED("engine_forkchoiceUpdated"),
+  ENGINE_GET_BLOBS("engine_getBlobs");
 
   private final String name;
 

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/methods/EngineGetBlobsV1.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright Consensys Software Inc., 2024
+ *
+ * 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.ethereum.executionclient.methods;
+
+import java.util.List;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import tech.pegasys.teku.ethereum.executionclient.ExecutionEngineClient;
+import tech.pegasys.teku.ethereum.executionclient.response.ResponseUnwrapper;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.Spec;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSchema;
+import tech.pegasys.teku.spec.datastructures.execution.BlobAndProof;
+import tech.pegasys.teku.spec.logic.versions.deneb.types.VersionedHash;
+import tech.pegasys.teku.spec.schemas.SchemaDefinitions;
+import tech.pegasys.teku.spec.schemas.SchemaDefinitionsDeneb;
+
+public class EngineGetBlobsV1 extends AbstractEngineJsonRpcMethod<List<BlobAndProof>> {
+
+  private static final Logger LOG = LogManager.getLogger();
+  private final Spec spec;
+
+  public EngineGetBlobsV1(final ExecutionEngineClient executionEngineClient, final Spec spec) {
+    super(executionEngineClient);
+    this.spec = spec;
+  }
+
+  @Override
+  public String getName() {
+    return EngineApiMethod.ENGINE_GET_BLOBS.getName();
+  }
+
+  @Override
+  public int getVersion() {
+    return 1;
+  }
+
+  @Override
+  public SafeFuture<List<BlobAndProof>> execute(final JsonRpcRequestParams params) {
+
+    final List<VersionedHash> blobVersionedHashes =
+        params.getRequiredListParameter(0, VersionedHash.class);
+
+    final UInt64 slot = params.getRequiredParameter(1, UInt64.class);
+
+    LOG.trace(
+        "Calling {}(blobVersionedHashes={}, slot={})",
+        getVersionedName(),
+        blobVersionedHashes,
+        slot);
+
+    return executionEngineClient
+        .getBlobsV1(blobVersionedHashes)
+        .thenApply(ResponseUnwrapper::unwrapExecutionClientResponseOrThrow)
+        .thenApply(
+            response -> {
+              final SchemaDefinitions schemaDefinitions = spec.atSlot(slot).getSchemaDefinitions();
+              final BlobSchema blobSchema =
+                  SchemaDefinitionsDeneb.required(schemaDefinitions).getBlobSchema();
+              return response.stream()
+                  .map(
+                      blobAndProofV1 ->
+                          blobAndProofV1 == null
+                              ? null
+                              : blobAndProofV1.asInternalBlobsAndProofs(blobSchema))
+                  .toList();
+            })
+        .thenPeek(
+            blobsAndProofs ->
+                LOG.trace(
+                    "Response {}(blobVersionedHashes={}) -> {}",
+                    getVersionedName(),
+                    blobVersionedHashes,
+                    blobsAndProofs));
+  }
+}

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/metrics/MetricRecordingExecutionEngineClient.java

@@ -19,6 +19,7 @@
 import org.apache.tuweni.bytes.Bytes32;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 import tech.pegasys.teku.ethereum.executionclient.ExecutionEngineClient;
+import tech.pegasys.teku.ethereum.executionclient.schema.BlobAndProofV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ClientVersionV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV2;
@@ -66,6 +67,7 @@ public class MetricRecordingExecutionEngineClient extends MetricRecordingAbstrac
   public static final String NEW_PAYLOAD_V4_METHOD = "new_payloadV4";
   public static final String EXCHANGE_CAPABILITIES_METHOD = "exchange_capabilities";
   public static final String GET_CLIENT_VERSION_V1_METHOD = "get_client_versionV1";
+  public static final String GET_BLOBS_V1_METHOD = "get_blobs_versionV1";
 
   private final ExecutionEngineClient delegate;
 
@@ -192,4 +194,10 @@ public SafeFuture<Response<List<ClientVersionV1>>> getClientVersionV1(
     return countRequest(
         () -> delegate.getClientVersionV1(clientVersion), GET_CLIENT_VERSION_V1_METHOD);
   }
+
+  @Override
+  public SafeFuture<Response<List<BlobAndProofV1>>> getBlobsV1(
+      final List<VersionedHash> blobVersionedHashes) {
+    return countRequest(() -> delegate.getBlobsV1(blobVersionedHashes), GET_BLOBS_V1_METHOD);
+  }
 }

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/schema/BlobAndProofV1.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright Consensys Software Inc., 2024
+ *
+ * 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.ethereum.executionclient.schema;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+import org.apache.tuweni.bytes.Bytes;
+import org.apache.tuweni.bytes.Bytes48;
+import tech.pegasys.teku.ethereum.executionclient.serialization.Bytes48Deserializer;
+import tech.pegasys.teku.ethereum.executionclient.serialization.BytesDeserializer;
+import tech.pegasys.teku.ethereum.executionclient.serialization.BytesSerializer;
+import tech.pegasys.teku.kzg.KZGProof;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.Blob;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSchema;
+import tech.pegasys.teku.spec.datastructures.execution.BlobAndProof;
+
+public class BlobAndProofV1 {
+
+  @JsonSerialize(using = BytesSerializer.class)
+  @JsonDeserialize(using = BytesDeserializer.class)
+  private final Bytes blob;
+
+  @JsonSerialize(using = BytesSerializer.class)
+  @JsonDeserialize(using = Bytes48Deserializer.class)
+  private final Bytes48 proof;
+
+  public BlobAndProofV1(
+      @JsonProperty("blob") final Bytes blob, @JsonProperty("proof") final Bytes48 proof) {
+    checkNotNull(blob, "blob");
+    checkNotNull(proof, "proof");
+    this.proof = proof;
+    this.blob = blob;
+  }
+
+  public BlobAndProof asInternalBlobsAndProofs(final BlobSchema blobSchema) {
+    return new BlobAndProof(new Blob(blobSchema, blob), new KZGProof(proof));
+  }
+
+  public static BlobAndProofV1 fromInternalBlobsBundle(final BlobAndProof blobAndProof) {
+    return new BlobAndProofV1(
+        blobAndProof.blob().getBytes(), blobAndProof.proof().getBytesCompressed());
+  }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final BlobAndProofV1 that = (BlobAndProofV1) o;
+    return Objects.equals(blob, that.blob) && Objects.equals(proof, that.proof);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(blob, proof);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("blob", bytesToBriefString(blob))
+        .add("proof", bytesToBriefString(proof))
+        .toString();
+  }
+
+  private String bytesToBriefString(final Bytes bytes) {
+    return bytes.slice(0, 7).toUnprefixedHexString();
+  }
+}

ethereum/executionclient/src/main/java/tech/pegasys/teku/ethereum/executionclient/web3j/Web3JExecutionEngineClient.java

@@ -27,6 +27,7 @@
 import org.web3j.protocol.core.Request;
 import org.web3j.protocol.core.methods.response.EthBlock;
 import tech.pegasys.teku.ethereum.executionclient.ExecutionEngineClient;
+import tech.pegasys.teku.ethereum.executionclient.schema.BlobAndProofV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ClientVersionV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV1;
 import tech.pegasys.teku.ethereum.executionclient.schema.ExecutionPayloadV2;
@@ -52,6 +53,7 @@ public class Web3JExecutionEngineClient implements ExecutionEngineClient {
 
   private static final Duration EXCHANGE_CAPABILITIES_TIMEOUT = Duration.ofSeconds(1);
   private static final Duration GET_CLIENT_VERSION_TIMEOUT = Duration.ofSeconds(1);
+  private static final Duration GET_BLOBS_TIMEOUT = Duration.ofSeconds(1);
 
   private final Web3JClient web3JClient;
 
@@ -253,6 +255,20 @@ public SafeFuture<Response<List<ClientVersionV1>>> getClientVersionV1(
     return web3JClient.doRequest(web3jRequest, GET_CLIENT_VERSION_TIMEOUT);
   }
 
+  @Override
+  public SafeFuture<Response<List<BlobAndProofV1>>> getBlobsV1(
+      final List<VersionedHash> blobVersionedHashes) {
+    final List<String> expectedBlobVersionedHashes =
+        blobVersionedHashes.stream().map(VersionedHash::toHexString).toList();
+    final Request<?, GetBlobsVersionV1Web3jResponse> web3jRequest =
+        new Request<>(
+            "engine_getBlobsV1",
+            list(expectedBlobVersionedHashes),
+            web3JClient.getWeb3jService(),
+            GetBlobsVersionV1Web3jResponse.class);
+    return web3JClient.doRequest(web3jRequest, GET_BLOBS_TIMEOUT);
+  }
+
   static class ExecutionPayloadV1Web3jResponse
       extends org.web3j.protocol.core.Response<ExecutionPayloadV1> {}
 
@@ -277,6 +293,9 @@ static class ExchangeCapabilitiesWeb3jResponse
   static class GetClientVersionV1Web3jResponse
       extends org.web3j.protocol.core.Response<List<ClientVersionV1>> {}
 
+  static class GetBlobsVersionV1Web3jResponse
+      extends org.web3j.protocol.core.Response<List<BlobAndProofV1>> {}
+
   /**
    * Returns a list that supports null items.
    *