Blob sidecars availability checker (#7081)

Consensys · May 5, 2023 · 5795062 · 5795062
1 parent 7a273fd
commit 5795062
Show file tree

Hide file tree

Showing 34 changed files with 1,186 additions and 433 deletions.
diff --git a/...n/sync/src/main/java/tech/pegasys/teku/beacon/sync/historical/HistoricalBatchFetcher.java b/...n/sync/src/main/java/tech/pegasys/teku/beacon/sync/historical/HistoricalBatchFetcher.java
@@ -53,6 +53,7 @@
 import tech.pegasys.teku.spec.datastructures.state.Fork;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
 import tech.pegasys.teku.spec.logic.common.util.AsyncBLSSignatureVerifier;
+import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobSidecarsAndValidationResult;
 import tech.pegasys.teku.statetransition.blobs.BlobSidecarManager;
 import tech.pegasys.teku.storage.api.StorageUpdateChannel;
 import tech.pegasys.teku.storage.client.CombinedChainDataClient;
@@ -328,10 +329,8 @@ private SafeFuture<Void> importBatch() {
         .thenCompose(
             __ -> {
               final UInt64 latestSlotInBatch = blocksToImport.getLast().getSlot();
-              return validateBlobSidecars(latestSlotInBatch, blocksToImport);
-            })
-        .thenCompose(
-            validSignatures -> {
+              validateBlobSidecars(latestSlotInBatch, blocksToImport);
+
               final SignedBeaconBlock newEarliestBlock = blocksToImport.getFirst();
               return storageUpdateChannel
                   .onFinalizedBlocks(blocksToImport, new HashMap<>(blobSidecarsBySlotToImport))
@@ -392,36 +391,33 @@ private SafeFuture<Void> batchVerifyHistoricalBlockSignature(
             });
   }
 
-  private SafeFuture<Void> validateBlobSidecars(
+  private void validateBlobSidecars(
       final UInt64 latestSlotInBatch, final Collection<SignedBeaconBlock> blocks) {
     if (!blobSidecarManager.isAvailabilityRequiredAtSlot(latestSlotInBatch)) {
-      return SafeFuture.COMPLETE;
+      return;
     }
     LOG.trace("Validating blob sidecars for a batch");
-    return SafeFuture.allOfFailFast(blocks.stream().map(this::validateBlobSidecars));
+    blocks.forEach(this::validateBlobSidecars);
   }
 
-  private SafeFuture<Void> validateBlobSidecars(final SignedBeaconBlock block) {
+  private void validateBlobSidecars(final SignedBeaconBlock block) {
     final List<BlobSidecar> blobSidecars =
         blobSidecarsBySlotToImport.getOrDefault(block.getSlot(), Collections.emptyList());
     LOG.trace("Validating {} blob sidecars for block {}", blobSidecars.size(), block.getRoot());
-    return blobSidecarManager
-        .createAvailabilityChecker(block)
-        .validate(blobSidecars)
-        .thenAccept(
-            validationResult -> {
-              if (validationResult.isFailure()) {
-                final String causeMessage =
-                    validationResult
-                        .getCause()
-                        .map(cause -> " (" + ExceptionUtil.getRootCauseMessage(cause) + ")")
-                        .orElse("");
-                throw new IllegalArgumentException(
-                    String.format(
-                        "Blob sidecars validation for block %s failed: %s%s",
-                        block.getRoot(), validationResult.getValidationResult(), causeMessage));
-              }
-            });
+    final BlobSidecarsAndValidationResult validationResult =
+        blobSidecarManager.createAvailabilityChecker(block).validateImmediately(blobSidecars);
+
+    if (validationResult.isFailure()) {
+      final String causeMessage =
+          validationResult
+              .getCause()
+              .map(cause -> " (" + ExceptionUtil.getRootCauseMessage(cause) + ")")
+              .orElse("");
+      throw new IllegalArgumentException(
+          String.format(
+              "Blob sidecars validation for block %s failed: %s%s",
+              block.getRoot(), validationResult.getValidationResult(), causeMessage));
+    }
   }
 
   private RequestParameters calculateRequestParams() {

diff --git a/...nc/src/test/java/tech/pegasys/teku/beacon/sync/historical/HistoricalBatchFetcherTest.java b/...nc/src/test/java/tech/pegasys/teku/beacon/sync/historical/HistoricalBatchFetcherTest.java
@@ -146,11 +146,8 @@ public void setup() {
         .thenReturn(SafeFuture.completedFuture(true));
     when(blobSidecarManager.createAvailabilityChecker(any()))
         .thenReturn(blobSidecarsAvailabilityChecker);
-    when(blobSidecarsAvailabilityChecker.validate(anyList()))
-        .thenAnswer(
-            i ->
-                SafeFuture.completedFuture(
-                    BlobSidecarsAndValidationResult.validResult(i.getArgument(0))));
+    when(blobSidecarsAvailabilityChecker.validateImmediately(anyList()))
+        .thenAnswer(i -> BlobSidecarsAndValidationResult.validResult(i.getArgument(0)));
   }
 
   @Test
@@ -201,12 +198,11 @@ public void run_returnAllBlocksAndBlobSidecarsOnFirstRequest() {
   @Test
   public void run_failsOnBlobSidecarsValidationFailure() {
     when(blobSidecarManager.isAvailabilityRequiredAtSlot(any())).thenReturn(true);
-    when(blobSidecarsAvailabilityChecker.validate(anyList()))
+    when(blobSidecarsAvailabilityChecker.validateImmediately(anyList()))
         .thenAnswer(
             i ->
-                SafeFuture.completedFuture(
-                    BlobSidecarsAndValidationResult.invalidResult(
-                        i.getArgument(0), new IllegalStateException("oopsy"))));
+                BlobSidecarsAndValidationResult.invalidResult(
+                    i.getArgument(0), new IllegalStateException("oopsy")));
 
     assertThat(peer.getOutstandingRequests()).isEqualTo(0);
     final SafeFuture<BeaconBlockSummary> future = fetcher.run();

diff --git a/beacon/sync/src/testFixtures/java/tech/pegasys/teku/beacon/sync/SyncingNodeManager.java b/beacon/sync/src/testFixtures/java/tech/pegasys/teku/beacon/sync/SyncingNodeManager.java
@@ -147,6 +147,7 @@ public static SyncingNodeManager create(
         new BlockManager(
             recentChainData,
             blockImporter,
+            blobSidecarPool,
             pendingBlocks,
             futureBlocks,
             invalidBlockRoots,

diff --git a/...rc/referenceTest/java/tech/pegasys/teku/reference/altair/fork/TransitionTestExecutor.java b/...rc/referenceTest/java/tech/pegasys/teku/reference/altair/fork/TransitionTestExecutor.java
@@ -32,7 +32,6 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.StateTransitionException;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 
 public class TransitionTestExecutor implements TestExecutor {
@@ -102,12 +101,7 @@ private void processUpgrade(final TestDefinition testDefinition, final MetaData
             metadata.blsSetting == 2 ? BLSSignatureVerifier.NO_OP : BLSSignatureVerifier.SIMPLE;
         result =
             spec.processBlock(
-                result,
-                block,
-                signatureVerifier,
-                Optional.empty(),
-                KzgCommitmentsProcessor.NOOP,
-                BlobsSidecarAvailabilityChecker.NOOP);
+                result, block, signatureVerifier, Optional.empty(), KzgCommitmentsProcessor.NOOP);
       } catch (final StateTransitionException e) {
         Assertions.fail(
             "Failed to process block " + i + " at slot " + block.getSlot() + ": " + e.getMessage(),

diff --git a/...eferenceTest/java/tech/pegasys/teku/reference/phase0/sanity/SanityBlocksTestExecutor.java b/...eferenceTest/java/tech/pegasys/teku/reference/phase0/sanity/SanityBlocksTestExecutor.java
@@ -34,7 +34,6 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.StateTransitionException;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 import tech.pegasys.teku.spec.logic.versions.deneb.helpers.MiscHelpersDeneb;
 
@@ -125,8 +124,7 @@ private BeaconState applyBlocks(
                     ? BLSSignatureVerifier.NO_OP
                     : BLSSignatureVerifier.SIMPLE,
                 Optional.empty(),
-                kzgCommitmentsProcessor,
-                BlobsSidecarAvailabilityChecker.NOOP);
+                kzgCommitmentsProcessor);
       }
       return result;
     } catch (StateTransitionException e) {

diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/Spec.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/Spec.java
@@ -96,7 +96,6 @@
 import tech.pegasys.teku.spec.logic.common.util.LightClientUtil;
 import tech.pegasys.teku.spec.logic.common.util.SyncCommitteeUtil;
 import tech.pegasys.teku.spec.logic.versions.bellatrix.block.OptimisticExecutionPayloadExecutor;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 import tech.pegasys.teku.spec.schemas.SchemaDefinitions;
 
@@ -701,8 +700,7 @@ public BeaconState processBlock(
       final SignedBeaconBlock block,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws StateTransitionException {
     try {
       final BeaconState blockSlotState = stateTransition.processSlots(preState, block.getSlot());
@@ -713,8 +711,7 @@ public BeaconState processBlock(
               IndexedAttestationCache.NOOP,
               signatureVerifier,
               payloadExecutor,
-              kzgCommitmentsProcessor,
-              blobsSidecarAvailabilityChecker);
+              kzgCommitmentsProcessor);
     } catch (SlotProcessingException | EpochProcessingException e) {
       throw new StateTransitionException(e);
     }
@@ -731,8 +728,7 @@ public BeaconState replayValidatedBlock(final BeaconState preState, final Signed
               IndexedAttestationCache.NOOP,
               BLSSignatureVerifier.NO_OP,
               Optional.empty(),
-              KzgCommitmentsProcessor.NOOP,
-              BlobsSidecarAvailabilityChecker.NOOP);
+              KzgCommitmentsProcessor.NOOP);
     } catch (SlotProcessingException | EpochProcessingException | BlockProcessingException e) {
       throw new StateTransitionException(e);
     }

diff --git a/.../spec/src/main/java/tech/pegasys/teku/spec/logic/common/block/AbstractBlockProcessor.java b/.../spec/src/main/java/tech/pegasys/teku/spec/logic/common/block/AbstractBlockProcessor.java
@@ -77,7 +77,6 @@
 import tech.pegasys.teku.spec.logic.common.util.BeaconStateUtil;
 import tech.pegasys.teku.spec.logic.common.util.ValidatorsUtil;
 import tech.pegasys.teku.spec.logic.versions.bellatrix.block.OptimisticExecutionPayloadExecutor;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 
 public abstract class AbstractBlockProcessor implements BlockProcessor {
@@ -135,8 +134,7 @@ public BeaconState processAndValidateBlock(
       final BeaconState blockSlotState,
       final IndexedAttestationCache indexedAttestationCache,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws StateTransitionException {
     final BatchSignatureVerifier signatureVerifier = new BatchSignatureVerifier();
     final BeaconState result =
@@ -146,8 +144,7 @@ public BeaconState processAndValidateBlock(
             indexedAttestationCache,
             signatureVerifier,
             payloadExecutor,
-            kzgCommitmentsProcessor,
-            blobsSidecarAvailabilityChecker);
+            kzgCommitmentsProcessor);
     if (!signatureVerifier.batchVerify()) {
       throw new StateTransitionException(
           "Batch signature verification failed for block " + signedBlock.toLogString());
@@ -162,8 +159,7 @@ public BeaconState processAndValidateBlock(
       final IndexedAttestationCache indexedAttestationCache,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws StateTransitionException {
     try {
       final BlockValidationResult preValidationResult =
@@ -180,8 +176,7 @@ public BeaconState processAndValidateBlock(
               indexedAttestationCache,
               signatureVerifier,
               payloadExecutor,
-              kzgCommitmentsProcessor,
-              blobsSidecarAvailabilityChecker);
+              kzgCommitmentsProcessor);
 
       BlockValidationResult blockValidationResult =
           validateBlockPostProcessing(
@@ -326,8 +321,7 @@ public BeaconState processUnsignedBlock(
       final IndexedAttestationCache indexedAttestationCache,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws BlockProcessingException {
     return preState.updated(
         state ->
@@ -337,8 +331,7 @@ public BeaconState processUnsignedBlock(
                 indexedAttestationCache,
                 signatureVerifier,
                 payloadExecutor,
-                kzgCommitmentsProcessor,
-                blobsSidecarAvailabilityChecker));
+                kzgCommitmentsProcessor));
   }
 
   protected void processBlock(
@@ -347,8 +340,7 @@ protected void processBlock(
       final IndexedAttestationCache indexedAttestationCache,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws BlockProcessingException {
     processBlockHeader(state, block);
     processRandaoNoValidation(state, block.getBody());

diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/block/BlockProcessor.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/block/BlockProcessor.java
@@ -47,7 +47,6 @@
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.BlockProcessingException;
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.StateTransitionException;
 import tech.pegasys.teku.spec.logic.versions.bellatrix.block.OptimisticExecutionPayloadExecutor;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 
 public interface BlockProcessor {
@@ -59,8 +58,7 @@ BeaconState processAndValidateBlock(
       BeaconState blockSlotState,
       IndexedAttestationCache indexedAttestationCache,
       Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws StateTransitionException;
 
   /**
@@ -81,8 +79,7 @@ BeaconState processAndValidateBlock(
       IndexedAttestationCache indexedAttestationCache,
       BLSSignatureVerifier signatureVerifier,
       Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws StateTransitionException;
 
   BeaconState processUnsignedBlock(
@@ -91,8 +88,7 @@ BeaconState processUnsignedBlock(
       IndexedAttestationCache indexedAttestationCache,
       BLSSignatureVerifier signatureVerifier,
       Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws BlockProcessingException;
 
   void processBlockHeader(MutableBeaconState state, BeaconBlockSummary blockHeader)

diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/util/BlockProposalUtil.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/util/BlockProposalUtil.java
@@ -32,7 +32,6 @@
 import tech.pegasys.teku.spec.logic.common.block.BlockProcessor;
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.BlockProcessingException;
 import tech.pegasys.teku.spec.logic.common.statetransition.exceptions.StateTransitionException;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 import tech.pegasys.teku.spec.schemas.SchemaDefinitions;
 
@@ -97,8 +96,7 @@ public SafeFuture<BeaconBlockAndState> createNewUnsignedBlock(
                       IndexedAttestationCache.NOOP,
                       BLSSignatureVerifier.NO_OP,
                       Optional.empty(),
-                      KzgCommitmentsProcessor.NOOP,
-                      BlobsSidecarAvailabilityChecker.NOOP);
+                      KzgCommitmentsProcessor.NOOP);
 
               final Bytes32 stateRoot = newState.hashTreeRoot();
               final BeaconBlock newCompleteBlock = block.withStateRoot(stateRoot);

diff --git a/...rc/main/java/tech/pegasys/teku/spec/logic/versions/altair/block/BlockProcessorAltair.java b/...rc/main/java/tech/pegasys/teku/spec/logic/versions/altair/block/BlockProcessorAltair.java
@@ -67,7 +67,6 @@
 import tech.pegasys.teku.spec.logic.versions.altair.helpers.BeaconStateAccessorsAltair;
 import tech.pegasys.teku.spec.logic.versions.altair.helpers.MiscHelpersAltair;
 import tech.pegasys.teku.spec.logic.versions.bellatrix.block.OptimisticExecutionPayloadExecutor;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 
 public class BlockProcessorAltair extends AbstractBlockProcessor {
@@ -113,8 +112,7 @@ public void processBlock(
       final IndexedAttestationCache indexedAttestationCache,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws BlockProcessingException {
     final MutableBeaconStateAltair state = MutableBeaconStateAltair.required(genericState);
     final BeaconBlockBodyAltair blockBody = BeaconBlockBodyAltair.required(block.getBody());
@@ -125,8 +123,7 @@ public void processBlock(
         indexedAttestationCache,
         signatureVerifier,
         payloadExecutor,
-        kzgCommitmentsProcessor,
-        blobsSidecarAvailabilityChecker);
+        kzgCommitmentsProcessor);
     processSyncAggregate(state, blockBody.getSyncAggregate(), signatureVerifier);
   }
 

diff --git a/...n/java/tech/pegasys/teku/spec/logic/versions/bellatrix/block/BlockProcessorBellatrix.java b/...n/java/tech/pegasys/teku/spec/logic/versions/bellatrix/block/BlockProcessorBellatrix.java
@@ -42,7 +42,6 @@
 import tech.pegasys.teku.spec.logic.versions.altair.block.BlockProcessorAltair;
 import tech.pegasys.teku.spec.logic.versions.altair.helpers.BeaconStateAccessorsAltair;
 import tech.pegasys.teku.spec.logic.versions.bellatrix.helpers.MiscHelpersBellatrix;
-import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobsSidecarAvailabilityChecker;
 import tech.pegasys.teku.spec.logic.versions.deneb.block.KzgCommitmentsProcessor;
 import tech.pegasys.teku.spec.schemas.SchemaDefinitionsBellatrix;
 
@@ -86,8 +85,7 @@ public void processBlock(
       final IndexedAttestationCache indexedAttestationCache,
       final BLSSignatureVerifier signatureVerifier,
       final Optional<? extends OptimisticExecutionPayloadExecutor> payloadExecutor,
-      final KzgCommitmentsProcessor kzgCommitmentsProcessor,
-      final BlobsSidecarAvailabilityChecker blobsSidecarAvailabilityChecker)
+      final KzgCommitmentsProcessor kzgCommitmentsProcessor)
       throws BlockProcessingException {
     final MutableBeaconStateBellatrix state = MutableBeaconStateBellatrix.required(genericState);
     final BeaconBlockBody blockBody = block.getBody();