feat(tests,spec): implement eip-8070 sparse blob pool infra

packages/testing/src/execution_testing/execution/blob_transaction.py

@@ -14,11 +14,16 @@
 from execution_testing.rpc import (
     BlobAndProofV1,
     BlobAndProofV2,
+    BlobCellsAndProofsV1,
     EngineRPC,
     EthRPC,
 )
-from execution_testing.rpc.rpc_types import GetBlobsResponse
+from execution_testing.rpc.rpc_types import (
+    GetBlobsResponse,
+    GetBlobsV4Response,
+)
 from execution_testing.test_types import (
+    Blob,
     NetworkWrappedTransaction,
     Transaction,
 )
@@ -106,6 +111,80 @@ def _validate_blob_and_proof(
         )
 
 
+def _validate_cells_and_proofs(
+    expected_blob: Blob | None,
+    received: BlobCellsAndProofsV1 | None,
+    cell_mask: int,
+    index: int,
+) -> None:
+    """
+    Validate an `engine_getBlobsV4` partial cell matrix against a local blob.
+
+    For each cell index `i`, the bit at position `i` in `cell_mask` decides
+    whether the cell was requested. Requested cells (and their proofs) must
+    match the locally computed values; non-requested cells must be `null`.
+
+    When `expected_blob` is `None` (a non-existing versioned hash), the whole
+    entry must be `null`.
+    """
+    if expected_blob is None:
+        if received is None:
+            logger.info(
+                f"Blob at index {index} correctly returned null "
+                "(non-existing blob hash)"
+            )
+            return
+        raise ValueError(
+            f"Blob at index {index} should not exist but "
+            f"client returned a cell matrix."
+        )
+    if received is None:
+        raise ValueError(f"Received cell matrix at index {index} is empty.")
+
+    assert expected_blob.cells is not None, (
+        "Local blob has no cells; getBlobsV4 requires a fork with cell proofs."
+    )
+    assert isinstance(expected_blob.proof, list), (
+        "Local blob proof is not a cell-proof list."
+    )
+    cells_per_ext_blob = len(expected_blob.cells)
+    if len(received.blob_cells) != cells_per_ext_blob:
+        raise ValueError(
+            f"Cell matrix at index {index} has {len(received.blob_cells)} "
+            f"cells, expected {cells_per_ext_blob}."
+        )
+    if len(received.proofs) != cells_per_ext_blob:
+        raise ValueError(
+            f"Proof matrix at index {index} has {len(received.proofs)} "
+            f"proofs, expected {cells_per_ext_blob}."
+        )
+
+    for i in range(cells_per_ext_blob):
+        requested = (cell_mask >> i) & 1
+        recv_cell = received.blob_cells[i]
+        recv_proof = received.proofs[i]
+        if requested:
+            if recv_cell != expected_blob.cells[i]:
+                raise ValueError(
+                    f"Cell mismatch at blob index {index}, cell {i}."
+                )
+            if recv_proof != expected_blob.proof[i]:
+                raise ValueError(
+                    f"Cell proof mismatch at blob index {index}, cell {i}."
+                )
+        else:
+            if recv_cell is not None:
+                raise ValueError(
+                    f"Cell at blob index {index}, cell {i} was not requested "
+                    "but client returned a non-null cell."
+                )
+            if recv_proof is not None:
+                raise ValueError(
+                    f"Proof at blob index {index}, cell {i} was not requested "
+                    "but client returned a non-null proof."
+                )
+
+
 def versioned_hashes_with_blobs_and_proofs(
     tx: NetworkWrappedTransaction,
 ) -> Dict[Hash, BlobAndProofV1 | BlobAndProofV2]:
@@ -148,6 +227,7 @@ class BlobTransaction(BaseExecute):
     txs: List[NetworkWrappedTransaction | Transaction]
     nonexisting_blob_hashes: List[Hash] | None = None
     get_blobs_version: int | None = None
+    cell_mask: int | None = None
 
     def get_required_sender_balances(
         self,
@@ -183,6 +263,7 @@ def execute(
     ) -> ExecuteResult:
         """Execute the format."""
         versioned_hashes: Dict[Hash, BlobAndProofV1 | BlobAndProofV2] = {}
+        blobs_by_hash: Dict[Hash, Blob] = {}
         sent_txs: List[Transaction] = []
         for tx_index, tx in enumerate(self.txs):
             tx = tx.with_signature_and_sender()
@@ -193,6 +274,8 @@ def execute(
                 versioned_hashes.update(
                     versioned_hashes_with_blobs_and_proofs(tx)
                 )
+                for blob in tx.blob_objects:
+                    blobs_by_hash[blob.versioned_hash] = blob
             else:
                 sent_txs.append(tx)
             label = (
@@ -234,12 +317,46 @@ def execute(
         if self.nonexisting_blob_hashes is not None:
             list_versioned_hashes.extend(self.nonexisting_blob_hashes)
 
-        blob_response: GetBlobsResponse | None = engine_rpc.get_blobs(
-            list_versioned_hashes, version=version
+        indices_bitarray = self.cell_mask if version >= 4 else None
+        blob_response: GetBlobsResponse | GetBlobsV4Response | None = (
+            engine_rpc.get_blobs(
+                list_versioned_hashes,
+                version=version,
+                indices_bitarray=indices_bitarray,
+            )
         )
 
-        if version <= 2:
+        if version >= 4:
+            # V4 (EIP-8070): partial cell matrix, selected by cell_mask
+            assert self.cell_mask is not None, (
+                f"getBlobsV{version} requires a cell_mask."
+            )
+            if blob_response is None:
+                raise ValueError(
+                    f"getBlobsV{version} returned 'null' for the entire "
+                    "response, but V4 should always return an array "
+                    "(with null entries for missing blobs)."
+                )
+            assert isinstance(blob_response, GetBlobsV4Response)
+            expected_blobs: List[Blob | None] = [
+                blobs_by_hash[vh] for vh in versioned_hashes
+            ]
+            if self.nonexisting_blob_hashes is not None:
+                expected_blobs += [None] * len(self.nonexisting_blob_hashes)
+            if len(blob_response) != len(expected_blobs):
+                raise ValueError(
+                    f"Expected {len(expected_blobs)} blob responses, "
+                    f"got {len(blob_response)}."
+                )
+            for i, (expected_cells, received_cells) in enumerate(
+                zip(expected_blobs, blob_response.root, strict=True)
+            ):
+                _validate_cells_and_proofs(
+                    expected_cells, received_cells, self.cell_mask, i
+                )
+        elif version <= 2:
             # V1/V2: all-or-nothing behavior
+            assert isinstance(blob_response, (GetBlobsResponse, type(None)))
             if self.nonexisting_blob_hashes is not None:
                 # Any missing blob must cause the entire response to be null
                 if blob_response is not None:
@@ -274,6 +391,7 @@ def execute(
                     _validate_blob_and_proof(expected_blob, received_blob, i)
         elif version == 3:
             # V3: partial responses (null only for missing blobs)
+            assert isinstance(blob_response, (GetBlobsResponse, type(None)))
             if blob_response is None:
                 raise ValueError(
                     f"getBlobsV{version} returned 'null' for the entire "
@@ -312,7 +430,7 @@ def execute(
         else:
             raise NotImplementedError(
                 f"getBlobsV{version} is not supported. "
-                "Supported versions: V1, V2, V3."
+                "Supported versions: V1, V2, V3, V4."
             )
 
         eth_rpc.wait_for_transactions(sent_txs)

packages/testing/src/execution_testing/rpc/__init__.py

@@ -20,6 +20,7 @@
 from .rpc_types import (
     BlobAndProofV1,
     BlobAndProofV2,
+    BlobCellsAndProofsV1,
     EthConfigResponse,
     ForkConfig,
     ForkConfigBlobSchedule,
@@ -33,6 +34,7 @@
     "AdminRPC",
     "BlobAndProofV1",
     "BlobAndProofV2",
+    "BlobCellsAndProofsV1",
     "BlockNotAvailableError",
     "BlockNumberType",
     "DebugRPC",

packages/testing/src/execution_testing/rpc/rpc.py

@@ -42,6 +42,7 @@
     ForkchoiceState,
     ForkchoiceUpdateResponse,
     GetBlobsResponse,
+    GetBlobsV4Response,
     GetPayloadResponse,
     JSONRPCRequest,
     JSONRPCResponse,
@@ -1314,21 +1315,35 @@ def get_blobs(
         versioned_hashes: List[Hash],
         *,
         version: int,
-    ) -> GetBlobsResponse | None:
+        indices_bitarray: int | None = None,
+    ) -> GetBlobsResponse | GetBlobsV4Response | None:
         """
         `engine_getBlobsVX`: Retrieves blobs from an execution layers tx pool.
+
+        For `V4` (EIP-8070), `indices_bitarray` is a required 16-byte (uint128)
+        cell mask selecting which cells to retrieve, encoded as a fixed-width
+        `DATA` value.
         """
         method = f"getBlobsV{version}"
-        params = [f"{h}" for h in versioned_hashes]
+        params: List[Any] = [[f"{h}" for h in versioned_hashes]]
+
+        if version >= 4:
+            assert indices_bitarray is not None, (
+                f"getBlobsV{version} requires an indices_bitarray cell mask."
+            )
+            params.append(f"0x{indices_bitarray.to_bytes(16, 'big').hex()}")
 
         response = self.post_request(
-            request=RPCCall(method=method, params=[params]),
+            request=RPCCall(method=method, params=params),
         ).result_or_raise()
         if response is None:  # for tests that request non-existing blobs
             logger.debug("get_blobs response received but it has value: None")
             return None
 
-        return GetBlobsResponse.model_validate(
+        response_model = (
+            GetBlobsV4Response if version >= 4 else GetBlobsResponse
+        )
+        return response_model.model_validate(
             response,
             context=self.response_validation_context,
         )

packages/testing/src/execution_testing/rpc/rpc_types.py

@@ -331,6 +331,36 @@ def __getitem__(
         return self.root[index]
 
 
+class BlobCellsAndProofsV1(CamelModel):
+    """
+    Represents a partial cell matrix for a single blob, as returned by
+    `engine_getBlobsV4` (EIP-8070: eth/72 - Sparse Blobpool).
+
+    `blob_cells` and `proofs` are partial matrices of length
+    `CELLS_PER_EXT_BLOB` (128), with `null` entries at indices the client was
+    not asked for (or does not hold).
+    """
+
+    blob_cells: List[Bytes | None]
+    proofs: List[Bytes | None]
+
+
+class GetBlobsV4Response(
+    EthereumTestRootModel[List[BlobCellsAndProofsV1 | None]]
+):
+    """Represents the response of an `engine_getBlobsV4` request."""
+
+    root: List[BlobCellsAndProofsV1 | None]
+
+    def __len__(self) -> int:
+        """Return the number of blob entries in the response."""
+        return len(self.root)
+
+    def __getitem__(self, index: int) -> BlobCellsAndProofsV1 | None:
+        """Return the blob cell matrix at the given index."""
+        return self.root[index]
+
+
 class ForkConfigBlobSchedule(CamelModel):
     """Representation of the blob schedule of a given fork."""
 

packages/testing/src/execution_testing/specs/blobs.py

@@ -24,6 +24,7 @@ class BlobsTest(BaseTest):
     txs: List[NetworkWrappedTransaction | Transaction]
     nonexisting_blob_hashes: List[Hash] | None = None
     get_blobs_version: int | None = None
+    cell_mask: int | None = None
 
     supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
         LabeledExecuteFormat(
@@ -54,6 +55,7 @@ def execute(
                 txs=self.txs,
                 nonexisting_blob_hashes=self.nonexisting_blob_hashes,
                 get_blobs_version=self.get_blobs_version,
+                cell_mask=self.cell_mask,
             )
         raise Exception(f"Unsupported execute format: {execute_format}")
 

tests/amsterdam/eip8070_sparse_blobpool/__init__.py

@@ -0,0 +1,4 @@
+"""
+Test suite for
+[EIP-8070: eth/72 - Sparse Blobpool](https://eips.ethereum.org/EIPS/eip-8070).
+"""