feat: add transactionReceipts subscription support in eth_subscribe #…

CHANGELOG.md

@@ -41,6 +41,7 @@ are provided with different values, using input as per the execution-apis spec i
 - Upgrade besu-native libraries version to 1.5.0. This fixes the issue of besu-native/secp256r1 exporting OpenSSL symbols in JVM space. [besu-native #308](https://github.com/besu-eth/besu-native/pull/308)
 
 ### Additions and Improvements
+- Add `transactionReceipts` subscription type to `eth_subscribe` that pushes all transaction receipts when a new block is added, with optional `transactionHashes` filter to receive receipts for specific transactions only [#10190](https://github.com/besu-eth/besu/pull/10190)
 - `--net-restrict` now supports IPv6 CIDR notation (e.g. `fd00::/64`) in addition to IPv4, enabling subnet-based peer filtering in IPv6 and dual-stack deployments [#10028](https://github.com/besu-eth/besu/pull/10028)
 - Stop EngineQosTimer as part of shutdown [#9903](https://github.com/hyperledger/besu/pull/9903)
 - Add `--max-blobs-per-transaction` CLI option to configure the maximum number of blobs per transaction [#9912](https://github.com/hyperledger/besu/pull/9912)

app/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -61,6 +61,7 @@
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.pending.PendingTransactionDroppedSubscriptionService;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.pending.PendingTransactionSubscriptionService;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.syncing.SyncingSubscriptionService;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.transactionreceipts.TransactionReceiptsSubscriptionService;
 import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
@@ -934,6 +935,9 @@ public Runner build() {
           context.getBlockchain(), blockchainQueries, subscriptionManager);
 
       createSyncingSubscriptionService(synchronizer, subscriptionManager);
+
+      createTransactionReceiptsSubscriptionService(
+          context.getBlockchain(), blockchainQueries, subscriptionManager, protocolSchedule);
     }
 
     Optional<EngineJsonRpcService> engineJsonRpcService = Optional.empty();
@@ -1435,6 +1439,18 @@ private void createNewBlockHeadersSubscriptionService(
     blockchain.observeBlockAdded(newBlockHeadersSubscriptionService);
   }
 
+  private void createTransactionReceiptsSubscriptionService(
+      final Blockchain blockchain,
+      final BlockchainQueries blockchainQueries,
+      final SubscriptionManager subscriptionManager,
+      final ProtocolSchedule protocolSchedule) {
+    final TransactionReceiptsSubscriptionService transactionReceiptsSubscriptionService =
+        new TransactionReceiptsSubscriptionService(
+            subscriptionManager, blockchainQueries, protocolSchedule);
+
+    blockchain.observeBlockAdded(transactionReceiptsSubscriptionService);
+  }
+
   private WebSocketService createWebsocketService(
       final Vertx vertx,
       final WebSocketConfiguration configuration,

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/BlockReceiptsResult.java

@@ -19,7 +19,7 @@
 import com.fasterxml.jackson.annotation.JsonValue;
 
 /** The result set from querying the receipts for a given block. */
-public class BlockReceiptsResult {
+public class BlockReceiptsResult implements JsonRpcResult {
 
   private final List<TransactionReceiptResult> results;
 

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/SubscriptionBuilder.java

@@ -19,6 +19,7 @@
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.SubscribeRequest;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.SubscriptionType;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.syncing.SyncingSubscription;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.transactionreceipts.TransactionReceiptsSubscription;
 
 import java.util.function.Function;
 
@@ -41,6 +42,11 @@ public Subscription build(
         {
           return new SyncingSubscription(subscriptionId, connectionId, subscriptionType);
         }
+      case TRANSACTION_RECEIPTS:
+        {
+          return new TransactionReceiptsSubscription(
+              subscriptionId, connectionId, request.getTransactionHashes());
+        }
       case NEW_PENDING_TRANSACTIONS:
       default:
         return new Subscription(

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/request/SubscribeRequest.java

@@ -14,25 +14,38 @@
  */
 package org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request;
 
+import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.FilterParameter;
 
+import java.util.List;
 import java.util.Objects;
 
 public class SubscribeRequest {
 
   private final SubscriptionType subscriptionType;
   private final Boolean includeTransaction;
   private final FilterParameter filterParameter;
+  private final List<Hash> transactionHashes;
   private final String connectionId;
 
   public SubscribeRequest(
       final SubscriptionType subscriptionType,
       final FilterParameter filterParameter,
       final Boolean includeTransaction,
       final String connectionId) {
+    this(subscriptionType, filterParameter, includeTransaction, null, connectionId);
+  }
+
+  public SubscribeRequest(
+      final SubscriptionType subscriptionType,
+      final FilterParameter filterParameter,
+      final Boolean includeTransaction,
+      final List<Hash> transactionHashes,
+      final String connectionId) {
     this.subscriptionType = subscriptionType;
     this.includeTransaction = includeTransaction;
     this.filterParameter = filterParameter;
+    this.transactionHashes = transactionHashes;
     this.connectionId = connectionId;
   }
 
@@ -48,15 +61,19 @@ public Boolean getIncludeTransaction() {
     return includeTransaction;
   }
 
+  public List<Hash> getTransactionHashes() {
+    return transactionHashes;
+  }
+
   public String getConnectionId() {
     return this.connectionId;
   }
 
   @Override
   public String toString() {
     return String.format(
-        "SubscribeRequest{subscriptionType=%s, includeTransaction=%s, filterParameter=%s, connectionId=%s}",
-        subscriptionType, includeTransaction, filterParameter, connectionId);
+        "SubscribeRequest{subscriptionType=%s, includeTransaction=%s, filterParameter=%s, transactionHashes=%s, connectionId=%s}",
+        subscriptionType, includeTransaction, filterParameter, transactionHashes, connectionId);
   }
 
   @Override
@@ -71,11 +88,13 @@ public boolean equals(final Object o) {
     return subscriptionType == that.subscriptionType
         && Objects.equals(includeTransaction, that.includeTransaction)
         && Objects.equals(filterParameter, that.filterParameter)
+        && Objects.equals(transactionHashes, that.transactionHashes)
         && Objects.equals(connectionId, that.connectionId);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(subscriptionType, includeTransaction, filterParameter, connectionId);
+    return Objects.hash(
+        subscriptionType, includeTransaction, filterParameter, transactionHashes, connectionId);
   }
 }

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/request/SubscriptionRequestMapper.java

@@ -14,14 +14,17 @@
  */
 package org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request;
 
+import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.datatypes.parameters.UnsignedLongParameter;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.exception.InvalidJsonRpcParameters;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.FilterParameter;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.JsonRpcParameter.JsonRpcParameterException;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.RpcErrorType;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.methods.WebSocketRpcRequest;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.transactionreceipts.TransactionReceiptsFilterParameter;
 
+import java.util.List;
 import java.util.Optional;
 
 public class SubscriptionRequestMapper {
@@ -52,6 +55,10 @@ public SubscribeRequest mapSubscribeRequest(final JsonRpcRequestContext jsonRpcR
           {
             return parseLogsRequest(webSocketRpcRequestBody);
           }
+        case TRANSACTION_RECEIPTS:
+          {
+            return parseTransactionReceiptsRequest(webSocketRpcRequestBody);
+          }
         case NEW_PENDING_TRANSACTIONS:
         case SYNCING:
         default:
@@ -96,6 +103,22 @@ private SubscribeRequest parseLogsRequest(final WebSocketRpcRequest request) {
         SubscriptionType.LOGS, filterParameter, null, request.getConnectionId());
   }
 
+  private SubscribeRequest parseTransactionReceiptsRequest(final WebSocketRpcRequest request) {
+    final Optional<TransactionReceiptsFilterParameter> filterParam;
+    try {
+      filterParam = request.getOptionalParameter(1, TransactionReceiptsFilterParameter.class);
+    } catch (JsonRpcParameterException e) {
+      throw new InvalidJsonRpcParameters(
+          "Invalid transaction receipts filter parameter (index 1)",
+          RpcErrorType.INVALID_SUBSCRIPTION_PARAMS,
+          e);
+    }
+    final List<Hash> txHashes =
+        filterParam.map(TransactionReceiptsFilterParameter::getTransactionHashes).orElse(null);
+    return new SubscribeRequest(
+        SubscriptionType.TRANSACTION_RECEIPTS, null, null, txHashes, request.getConnectionId());
+  }
+
   public UnsubscribeRequest mapUnsubscribeRequest(final JsonRpcRequestContext jsonRpcRequestContext)
       throws InvalidSubscriptionRequestException {
     try {

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/request/SubscriptionType.java

@@ -30,7 +30,10 @@ public enum SubscriptionType {
   DROPPED_PENDING_TRANSACTIONS("droppedPendingTransactions"),
 
   @JsonProperty("syncing")
-  SYNCING("syncing");
+  SYNCING("syncing"),
+
+  @JsonProperty("transactionReceipts")
+  TRANSACTION_RECEIPTS("transactionReceipts");
 
   private final String code;
 

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/transactionreceipts/TransactionReceiptsFilterParameter.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright contributors to Besu.
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.transactionreceipts;
+
+import org.hyperledger.besu.datatypes.Hash;
+
+import java.util.Collections;
+import java.util.List;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class TransactionReceiptsFilterParameter {
+
+  private static final int MAX_TRANSACTION_HASHES = 200;
+  private final List<Hash> transactionHashes;
+
+  @JsonCreator
+  public TransactionReceiptsFilterParameter(
+      @JsonProperty("transactionHashes") final List<Hash> transactionHashes) {
+    if (transactionHashes != null && transactionHashes.size() > MAX_TRANSACTION_HASHES) {
+      throw new IllegalArgumentException(
+          "transactionHashes filter limited to " + MAX_TRANSACTION_HASHES + " entries");
+    }
+    this.transactionHashes =
+        transactionHashes == null ? Collections.emptyList() : transactionHashes;
+  }
+
+  public List<Hash> getTransactionHashes() {
+    return transactionHashes;
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/transactionreceipts/TransactionReceiptsSubscription.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright contributors to Besu.
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.transactionreceipts;
+
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.Subscription;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.SubscriptionType;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class TransactionReceiptsSubscription extends Subscription {
+
+  private final Set<Hash> transactionHashesFilter;
+
+  public TransactionReceiptsSubscription(
+      final Long subscriptionId, final String connectionId, final List<Hash> transactionHashes) {
+    super(subscriptionId, connectionId, SubscriptionType.TRANSACTION_RECEIPTS, Boolean.FALSE);
+    this.transactionHashesFilter =
+        transactionHashes == null || transactionHashes.isEmpty()
+            ? Collections.emptySet()
+            : new HashSet<>(transactionHashes);
+  }
+
+  public Set<Hash> getTransactionHashesFilter() {
+    return transactionHashesFilter;
+  }
+
+  public boolean hasFilter() {
+    return !transactionHashesFilter.isEmpty();
+  }
+}