[Stateless Proto] Introducing execution proofs subnet and gossip validation

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/executionproofs/ExecutionProofManager.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright Consensys Software Inc., 2025
+ *
+ * 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.statetransition.executionproofs;
+
+import java.util.Optional;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionProof;
+import tech.pegasys.teku.statetransition.blobs.RemoteOrigin;
+import tech.pegasys.teku.statetransition.validation.InternalValidationResult;
+
+public interface ExecutionProofManager {
+
+  ExecutionProofManager NOOP =
+      new ExecutionProofManager() {
+        @Override
+        public SafeFuture<InternalValidationResult> onExecutionProofGossip(
+            ExecutionProof executionProof, Optional<UInt64> arrivalTimestamp) {
+          return SafeFuture.completedFuture(InternalValidationResult.ACCEPT);
+        }
+
+        @Override
+        public void onExecutionProofPublish(
+            final ExecutionProof executionProof, final RemoteOrigin remoteOrigin) {}
+
+        @Override
+        public void subscribeToValidExecutionProofs(
+            final ValidExecutionProofListener sidecarsListener) {}
+      };
+
+  void onExecutionProofPublish(ExecutionProof executionProof, RemoteOrigin remoteOrigin);
+
+  SafeFuture<InternalValidationResult> onExecutionProofGossip(
+      ExecutionProof executionProof, Optional<UInt64> arrivalTimestamp);
+
+  void subscribeToValidExecutionProofs(
+      ExecutionProofManager.ValidExecutionProofListener executionProofListener);
+
+  interface ValidExecutionProofListener {
+    void onNewValidExecutionProof(ExecutionProof executionProof, RemoteOrigin remoteOrigin);
+  }
+}

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/executionproofs/ExecutionProofManagerImpl.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright Consensys Software Inc., 2025
+ *
+ * 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.statetransition.executionproofs;
+
+import java.util.Optional;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.subscribers.Subscribers;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionProof;
+import tech.pegasys.teku.statetransition.blobs.RemoteOrigin;
+import tech.pegasys.teku.statetransition.validation.ExecutionProofGossipValidator;
+import tech.pegasys.teku.statetransition.validation.InternalValidationResult;
+
+public class ExecutionProofManagerImpl implements ExecutionProofManager {
+
+  final ExecutionProofGossipValidator executionProofGossipValidator;
+  private final Subscribers<ValidExecutionProofListener> receivedExecutionProofSubscribers =
+      Subscribers.create(true);
+
+  private static final Logger LOG = LogManager.getLogger();
+
+  public ExecutionProofManagerImpl(
+      final ExecutionProofGossipValidator executionProofGossipValidator) {
+    this.executionProofGossipValidator = executionProofGossipValidator;
+  }
+
+  @Override
+  public void onExecutionProofPublish(
+      final ExecutionProof executionProof, final RemoteOrigin remoteOrigin) {
+    LOG.trace("Published execution proof {}", executionProof);
+  }
+
+  @Override
+  public SafeFuture<InternalValidationResult> onExecutionProofGossip(
+      final ExecutionProof executionProof, final Optional<UInt64> arrivalTimestamp) {
+    LOG.debug("Received execution proof for block {}", executionProof);
+    // TODO fix this as this needs to obtain subnetId from the topic we received the proof from?
+    // this is probably a validation that only makes sense for the stub we're doing
+    return executionProofGossipValidator.validate(
+        executionProof, executionProof.getSubnetId().get());
+  }
+
+  @Override
+  public void subscribeToValidExecutionProofs(
+      final ValidExecutionProofListener executionProofListener) {
+    receivedExecutionProofSubscribers.subscribe(executionProofListener);
+  }
+}

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright Consensys Software Inc., 2025
+ *
+ * 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.statetransition.validation;
+
+import static tech.pegasys.teku.spec.config.Constants.MAX_EXECUTION_PROOF_SUBNETS;
+
+import java.util.Set;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.collections.LimitedSet;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionProof;
+
+public class ExecutionProofGossipValidator {
+  private static final Logger LOG = LogManager.getLogger();
+
+  private final Set<ExecutionProof> receivedValidExecutionProofSet;
+
+  public static ExecutionProofGossipValidator create() {
+    return new ExecutionProofGossipValidator(
+        // max subnets * 2 epochs * slots per epoch 32 based on mainnet for now
+        LimitedSet.createSynchronized(MAX_EXECUTION_PROOF_SUBNETS.intValue() * 64));
+  }
+
+  public ExecutionProofGossipValidator(final Set<ExecutionProof> receivedValidExecutionProofSet) {
+
+    this.receivedValidExecutionProofSet = receivedValidExecutionProofSet;
+  }
+
+  public SafeFuture<InternalValidationResult> validate(
+      final ExecutionProof executionProof, final UInt64 subnetId) {
+
+    // TODO need to check for other validations done in the prototype and spec
+    if (!executionProof.getSubnetId().get().equals(subnetId)) {
+      LOG.trace(
+          "ExecutionProof for block root {} / block hash {} does not match the gossip subnetId",
+          executionProof.getBlockRoot(),
+          executionProof.getBlockHash());
+      return SafeFuture.completedFuture(InternalValidationResult.reject("SubnetId mismatch"));
+    }
+
+    // Already seen and valid
+    if (receivedValidExecutionProofSet.contains(executionProof)) {
+      LOG.trace("Received duplicate execution proof {}", executionProof);
+      return SafeFuture.completedFuture(InternalValidationResult.IGNORE);
+    }
+
+    // Validated the execution proof
+    LOG.trace(
+        "Received and validated execution proof for block root {}, block hash {}",
+        executionProof.getBlockRoot(),
+        executionProof.getBlockHash());
+    receivedValidExecutionProofSet.add(executionProof);
+    return SafeFuture.completedFuture(InternalValidationResult.ACCEPT);
+  }
+}

networking/eth2/build.gradle

@@ -15,6 +15,7 @@ dependencies {
   implementation project(':infrastructure:ssz')
   implementation project(':storage')
   implementation project(':storage:api')
+  implementation project(':services:zkchain')
   implementation project(':infrastructure:serviceutils')
 
   implementation 'io.libp2p:jvm-libp2p'

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/ActiveEth2P2PNetwork.java

@@ -30,6 +30,7 @@
 import tech.pegasys.teku.networking.eth2.gossip.BlobSidecarGossipChannel;
 import tech.pegasys.teku.networking.eth2.gossip.BlockGossipChannel;
 import tech.pegasys.teku.networking.eth2.gossip.DataColumnSidecarGossipChannel;
+import tech.pegasys.teku.networking.eth2.gossip.ExecutionProofGossipChannel;
 import tech.pegasys.teku.networking.eth2.gossip.config.Eth2Context;
 import tech.pegasys.teku.networking.eth2.gossip.config.GossipConfigurator;
 import tech.pegasys.teku.networking.eth2.gossip.encoding.GossipEncoding;
@@ -72,6 +73,7 @@ public class ActiveEth2P2PNetwork extends DelegatingP2PNetwork<Eth2Peer> impleme
   private final SubnetSubscriptionService attestationSubnetService;
   private final SubnetSubscriptionService syncCommitteeSubnetService;
   private final SubnetSubscriptionService dataColumnSidecarSubnetService;
+  private final SubnetSubscriptionService executionProofSubnetService;
   private final ProcessedAttestationSubscriptionProvider processedAttestationSubscriptionProvider;
   private final AtomicBoolean gossipStarted = new AtomicBoolean(false);
   private final int dasTotalCustodySubnetCount;
@@ -97,6 +99,7 @@ public ActiveEth2P2PNetwork(
       final SubnetSubscriptionService attestationSubnetService,
       final SubnetSubscriptionService syncCommitteeSubnetService,
       final SubnetSubscriptionService dataColumnSidecarSubnetService,
+      final SubnetSubscriptionService executionProofSubnetService,
       final GossipEncoding gossipEncoding,
       final GossipConfigurator gossipConfigurator,
       final ProcessedAttestationSubscriptionProvider processedAttestationSubscriptionProvider,
@@ -115,6 +118,7 @@ public ActiveEth2P2PNetwork(
     this.attestationSubnetService = attestationSubnetService;
     this.syncCommitteeSubnetService = syncCommitteeSubnetService;
     this.dataColumnSidecarSubnetService = dataColumnSidecarSubnetService;
+    this.executionProofSubnetService = executionProofSubnetService;
     this.processedAttestationSubscriptionProvider = processedAttestationSubscriptionProvider;
     this.dasTotalCustodySubnetCount = dasTotalCustodySubnetCount;
     this.allTopicsFilterEnabled = allTopicsFilterEnabled;
@@ -144,6 +148,9 @@ private synchronized void startup() {
     eventChannels.subscribe(
         DataColumnSidecarGossipChannel.class,
         (sidecar, __) -> gossipForkManager.publishDataColumnSidecar(sidecar));
+    eventChannels.subscribe(
+        ExecutionProofGossipChannel.class, gossipForkManager::publishExecutionProof);
+
     if (recentChainData.isCloseToInSync()) {
       startGossip();
     }
@@ -361,6 +368,18 @@ public void unsubscribeFromDataColumnSidecarSubnetId(final int subnetId) {
     dataColumnSidecarSubnetService.removeSubscription(subnetId);
   }
 
+  @Override
+  public void subscribeToExecutionProofSubnetId(final int subnetId) {
+    gossipForkManager.subscribeToExecutionProofSubnetId(subnetId);
+    executionProofSubnetService.addSubscription(subnetId);
+  }
+
+  @Override
+  public void unsubscribeFromExecutionProofSubnetId(final int subnetId) {
+    gossipForkManager.unsubscribeFromExecutionProofSubnetId(subnetId);
+    executionProofSubnetService.removeSubscription(subnetId);
+  }
+
   @Override
   public MetadataMessage getMetadata() {
     return peerManager.getMetadataMessage();

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/Eth2P2PNetwork.java

@@ -44,6 +44,10 @@ public interface Eth2P2PNetwork extends P2PNetwork<Eth2Peer> {
 
   void unsubscribeFromDataColumnSidecarSubnetId(int subnetId);
 
+  void subscribeToExecutionProofSubnetId(int subnetId);
+
+  void unsubscribeFromExecutionProofSubnetId(int subnetId);
+
   MetadataMessage getMetadata();
 
   void publishSyncCommitteeMessage(ValidatableSyncCommitteeMessage message);

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/Eth2P2PNetworkBuilder.java

@@ -86,6 +86,7 @@
 import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.spec.datastructures.epbs.versions.gloas.PayloadAttestationMessage;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionProof;
 import tech.pegasys.teku.spec.datastructures.operations.AttesterSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.ProposerSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.SignedBlsToExecutionChange;
@@ -148,6 +149,7 @@ public class Eth2P2PNetworkBuilder {
   protected OperationProcessor<ValidatableSyncCommitteeMessage>
       gossipedSyncCommitteeMessageProcessor;
   protected OperationProcessor<DataColumnSidecar> dataColumnSidecarOperationProcessor;
+  protected OperationProcessor<ExecutionProof> executionProofOperationProcessor;
   protected StatusMessageFactory statusMessageFactory;
   protected KZG kzg;
   protected boolean recordMessageArrival;
@@ -169,6 +171,7 @@ public Eth2P2PNetwork build() {
     final SubnetSubscriptionService syncCommitteeSubnetService = new SubnetSubscriptionService();
     final SubnetSubscriptionService dataColumnSidecarSubnetService =
         new SubnetSubscriptionService();
+    final SubnetSubscriptionService executionProofSubnetService = new SubnetSubscriptionService();
     final DiscoveryNodeIdExtractor discoveryNodeIdExtractor = new LibP2PDiscoveryNodeIdExtractor();
     final RpcEncoding rpcEncoding =
         RpcEncoding.createSszSnappyEncoding(spec.getNetworkingConfig().getMaxPayloadSize());
@@ -232,6 +235,7 @@ public Eth2P2PNetwork build() {
         attestationSubnetService,
         syncCommitteeSubnetService,
         dataColumnSidecarSubnetService,
+        executionProofSubnetService,
         gossipEncoding,
         config.getGossipConfigurator(),
         processedAttestationSubscriptionProvider,
@@ -381,7 +385,9 @@ private GossipForkSubscriptions createSubscriptions(
               gossipedSignedContributionAndProofProcessor,
               gossipedSyncCommitteeMessageProcessor,
               gossipedSignedBlsToExecutionChangeProcessor,
-              debugDataDumper);
+              debugDataDumper,
+              executionProofOperationProcessor,
+              config.isExecutionProofTopicEnabled());
       case FULU ->
           new GossipForkSubscriptionsFulu(
               forkAndSpecMilestone.getFork(),
@@ -403,7 +409,9 @@ private GossipForkSubscriptions createSubscriptions(
               gossipedSignedBlsToExecutionChangeProcessor,
               dataColumnSidecarOperationProcessor,
               debugDataDumper,
-              dasGossipLogger);
+              dasGossipLogger,
+              executionProofOperationProcessor,
+              config.isExecutionProofTopicEnabled());
       case GLOAS ->
           new GossipForkSubscriptionsGloas(
               forkAndSpecMilestone.getFork(),
@@ -426,7 +434,9 @@ private GossipForkSubscriptions createSubscriptions(
               dataColumnSidecarOperationProcessor,
               payloadAttestationMessageProcessor,
               debugDataDumper,
-              dasGossipLogger);
+              dasGossipLogger,
+              executionProofOperationProcessor,
+              config.isExecutionProofTopicEnabled());
     };
   }
 
@@ -456,9 +466,11 @@ private GossipForkSubscriptions createBpoSubscriptions(
               gossipedSyncCommitteeMessageProcessor,
               gossipedSignedBlsToExecutionChangeProcessor,
               dataColumnSidecarOperationProcessor,
+              executionProofOperationProcessor,
               debugDataDumper,
               dasGossipLogger,
-              bpo);
+              bpo,
+              config.isExecutionProofTopicEnabled());
       case GLOAS ->
           new GossipForkSubscriptionsGloasBpo(
               forkAndSpecMilestone.getFork(),
@@ -479,10 +491,12 @@ private GossipForkSubscriptions createBpoSubscriptions(
               gossipedSyncCommitteeMessageProcessor,
               gossipedSignedBlsToExecutionChangeProcessor,
               dataColumnSidecarOperationProcessor,
+              executionProofOperationProcessor,
               payloadAttestationMessageProcessor,
               debugDataDumper,
               dasGossipLogger,
-              bpo);
+              bpo,
+              config.isExecutionProofTopicEnabled());
       default ->
           throw new IllegalStateException(
               "BPO is not supported for: " + forkAndSpecMilestone.getSpecMilestone());
@@ -502,7 +516,7 @@ protected DiscoveryNetwork<?> buildNetwork(
             combinedChainDataClient.getRecentChainData()::getMilestoneByForkDigest);
     final GossipTopicFilter gossipTopicsFilter =
         new Eth2GossipTopicFilter(
-            combinedChainDataClient.getRecentChainData(), gossipEncoding, spec);
+            combinedChainDataClient.getRecentChainData(), gossipEncoding, spec, config);
     final NetworkConfig networkConfig = config.getNetworkConfig();
     final DiscoveryConfig discoConfig = config.getDiscoveryConfig();
 
@@ -615,6 +629,7 @@ private void validate() {
         "gossipedSignedBlsToExecutionChangeProcessor", gossipedSignedBlsToExecutionChangeProcessor);
     assertNotNull(
         "gossipedDataColumnSidecarOperationProcessor", dataColumnSidecarOperationProcessor);
+    // TODO maybe add an assertion here for EP Op Processor?
     assertNotNull("gossipedPayloadAttestationMessageProcessor", payloadAttestationMessageProcessor);
   }
 
@@ -755,6 +770,13 @@ public Eth2P2PNetworkBuilder gossipedDataColumnSidecarOperationProcessor(
     return this;
   }
 
+  public Eth2P2PNetworkBuilder gossipedExecutionProofOperationProcessor(
+      final OperationProcessor<ExecutionProof> executionProofOperationProcessor) {
+    checkNotNull(executionProofOperationProcessor);
+    this.executionProofOperationProcessor = executionProofOperationProcessor;
+    return this;
+  }
+
   public Eth2P2PNetworkBuilder gossipedPayloadAttestationMessageProcessor(
       final OperationProcessor<PayloadAttestationMessage>
           gossipedPayloadAttestationMessageProcessor) {