diff --git a/eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/deneb/merkle_proof/SingleMerkleProofTestExecutor.java b/eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/deneb/merkle_proof/SingleMerkleProofTestExecutor.java index 84e91f737bf..8186055167e 100644 --- a/eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/deneb/merkle_proof/SingleMerkleProofTestExecutor.java +++ b/eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/deneb/merkle_proof/SingleMerkleProofTestExecutor.java @@ -32,10 +32,13 @@ import tech.pegasys.teku.reference.TestDataUtils; import tech.pegasys.teku.reference.TestExecutor; import tech.pegasys.teku.spec.config.SpecConfigDeneb; +import tech.pegasys.teku.spec.config.SpecConfigFulu; import tech.pegasys.teku.spec.datastructures.blocks.blockbody.BeaconBlockBody; import tech.pegasys.teku.spec.datastructures.type.SszKZGCommitment; import tech.pegasys.teku.spec.logic.common.helpers.Predicates; import tech.pegasys.teku.spec.logic.versions.deneb.helpers.MiscHelpersDeneb; +import tech.pegasys.teku.spec.logic.versions.fulu.helpers.MiscHelpersFulu; +import tech.pegasys.teku.spec.schemas.SchemaDefinitionsFulu; public class SingleMerkleProofTestExecutor implements TestExecutor { private static final Pattern TEST_NAME_PATTERN = Pattern.compile("(.+)/(.+)"); @@ -88,9 +91,12 @@ void runBeaconBlockBodyTest( testDefinition, OBJECT_SSZ_FILE, testDefinition.getSpec().getGenesisSchemaDefinitions().getBeaconBlockBodySchema()); - + // Deneb if (proofType.startsWith("blob_kzg_commitment_merkle_proof")) { runBlobKzgCommitmentMerkleProofTest(testDefinition, data, beaconBlockBody); + // Fulu + } else if (proofType.startsWith("blob_kzg_commitments_merkle_proof")) { + runBlobKzgCommitmentsMerkleProofTest(testDefinition, data, beaconBlockBody); } else { throw new RuntimeException("Unknown proof type " + proofType); } @@ -130,11 +136,35 @@ private void runBlobKzgCommitmentMerkleProofTest( assertThat(miscHelpersDeneb.getBlobSidecarKzgCommitmentGeneralizedIndex(kzgCommitmentIndex)) .isEqualTo(data.leafIndex); assertThat( - miscHelpersDeneb.computeKzgCommitmentInclusionProof( + miscHelpersDeneb.computeBlobKzgCommitmentInclusionProof( kzgCommitmentIndex, beaconBlockBody)) .isEqualTo(data.branch.stream().map(Bytes32::fromHexString).toList()); } + private void runBlobKzgCommitmentsMerkleProofTest( + final TestDefinition testDefinition, final Data data, final BeaconBlockBody beaconBlockBody) { + final Predicates predicates = new Predicates(testDefinition.getSpec().getGenesisSpecConfig()); + final Bytes32 kzgCommitmentsHash = Bytes32.fromHexString(data.leaf); + + // Forward check + assertThat( + predicates.isValidMerkleBranch( + kzgCommitmentsHash, + createKzgCommitmentsMerkleProofBranchFromData(testDefinition, data.branch), + getKzgCommitmentsInclusionProofDepth(testDefinition), + data.leafIndex, + beaconBlockBody.hashTreeRoot())) + .isTrue(); + + // Verify 2 MiscHelpersFulu helpers + final MiscHelpersFulu miscHelpersFulu = + MiscHelpersFulu.required(testDefinition.getSpec().getGenesisSpec().miscHelpers()); + assertThat(miscHelpersFulu.getBlockBodyKzgCommitmentsGeneralizedIndex()) + .isEqualTo(data.leafIndex); + assertThat(miscHelpersFulu.computeDataColumnKzgCommitmentsInclusionProof(beaconBlockBody)) + .isEqualTo(data.branch.stream().map(Bytes32::fromHexString).toList()); + } + private SszBytes32Vector createKzgCommitmentMerkleProofBranchFromData( final TestDefinition testDefinition, final List branch) { final SszBytes32VectorSchema kzgCommitmentInclusionProofSchema = @@ -153,4 +183,20 @@ private int getKzgCommitmentInclusionProofDepth(final TestDefinition testDefinit return SpecConfigDeneb.required(testDefinition.getSpec().getGenesisSpecConfig()) .getKzgCommitmentInclusionProofDepth(); } + + private SszBytes32Vector createKzgCommitmentsMerkleProofBranchFromData( + final TestDefinition testDefinition, final List branch) { + final SszBytes32VectorSchema kzgCommitmentsInclusionProofSchema = + SchemaDefinitionsFulu.required(testDefinition.getSpec().getGenesisSchemaDefinitions()) + .getDataColumnSidecarSchema() + .getKzgCommitmentsInclusionProofSchema(); + return kzgCommitmentsInclusionProofSchema.createFromElements( + branch.stream().map(Bytes32::fromHexString).map(SszBytes32::of).toList()); + } + + private int getKzgCommitmentsInclusionProofDepth(final TestDefinition testDefinition) { + return SpecConfigFulu.required(testDefinition.getSpec().getGenesisSpecConfig()) + .getKzgCommitmentsInclusionProofDepth() + .intValue(); + } } 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 index cd2b98cee90..d0be794f9fb 100644 --- a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/Spec.java +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/Spec.java @@ -35,8 +35,6 @@ import java.util.function.Function; import java.util.function.Supplier; import javax.annotation.CheckReturnValue; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; import org.apache.tuweni.bytes.Bytes; import org.apache.tuweni.bytes.Bytes32; import tech.pegasys.teku.bls.BLSPublicKey; @@ -54,10 +52,12 @@ import tech.pegasys.teku.spec.config.SpecConfigAltair; import tech.pegasys.teku.spec.config.SpecConfigAndParent; import tech.pegasys.teku.spec.config.SpecConfigDeneb; +import tech.pegasys.teku.spec.config.SpecConfigFulu; import tech.pegasys.teku.spec.constants.Domain; import tech.pegasys.teku.spec.datastructures.attestation.ValidatableAttestation; import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.Blob; import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSidecar; import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlock; import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlockAndState; import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlockHeader; @@ -108,7 +108,6 @@ import tech.pegasys.teku.spec.schemas.registry.SchemaRegistryBuilder; public class Spec { - private static final Logger LOG = LogManager.getLogger(); private final Map specVersions; private final ForkSchedule forkSchedule; private final StateTransition stateTransition; @@ -147,13 +146,6 @@ static Spec create( final ForkSchedule forkSchedule = forkScheduleBuilder.build(); - final UInt64 lastForkActivationEpoch = - forkSchedule.getActiveMilestones().getLast().getFork().getEpoch(); - LOG.info( - "Creating network specification. Highest milestone supported: {}, from epoch {}", - highestMilestoneSupported.lowerCaseName(), - lastForkActivationEpoch); - return new Spec(specConfigAndParent, specVersions, forkSchedule); } @@ -433,6 +425,16 @@ public ExecutionPayloadHeader deserializeJsonExecutionPayloadHeader( .jsonDeserialize(objectMapper.createParser(jsonFile)); } + public DataColumnSidecar deserializeSidecar(final Bytes serializedSidecar, final UInt64 slot) { + return atSlot(slot) + .getSchemaDefinitions() + .toVersionFulu() + .orElseThrow( + () -> new RuntimeException("FULU milestone is required to deserialize column sidecar")) + .getDataColumnSidecarSchema() + .sszDeserialize(serializedSidecar); + } + // BeaconState public UInt64 getCurrentEpoch(final BeaconState state) { return atState(state).beaconStateAccessors().getCurrentEpoch(state); @@ -948,14 +950,14 @@ public boolean isAvailabilityOfBlobSidecarsRequiredAtSlot( public boolean isAvailabilityOfBlobSidecarsRequiredAtEpoch( final ReadOnlyStore store, final UInt64 epoch) { - if (!forkSchedule.getSpecMilestoneAtEpoch(epoch).isGreaterThanOrEqualTo(DENEB)) { - return false; - } - final SpecConfig config = atEpoch(epoch).getConfig(); - final SpecConfigDeneb specConfigDeneb = SpecConfigDeneb.required(config); - return getCurrentEpoch(store) - .minusMinZero(epoch) - .isLessThanOrEqualTo(specConfigDeneb.getMinEpochsForBlobSidecarsRequests()); + return atEpoch(epoch) + .miscHelpers() + .toVersionDeneb() + .map( + denebMiscHelpers -> + denebMiscHelpers.isAvailabilityOfBlobSidecarsRequiredAtEpoch( + getCurrentEpoch(store), epoch)) + .orElse(false); } /** @@ -1000,6 +1002,26 @@ public UInt64 computeSubnetForBlobSidecar(final BlobSidecar blobSidecar) { .getBlobSidecarSubnetCount()); } + public Optional getNumberOfDataColumns() { + return getSpecConfigFulu().map(SpecConfigFulu::getNumberOfColumns); + } + + public Optional getNumberOfDataColumnSubnets() { + return getSpecConfigFulu().map(SpecConfigFulu::getDataColumnSidecarSubnetCount); + } + + public boolean isAvailabilityOfDataColumnSidecarsRequiredAtEpoch( + final ReadOnlyStore store, final UInt64 epoch) { + if (getSpecConfigFulu().isEmpty()) { + return false; + } + final SpecConfig config = atEpoch(epoch).getConfig(); + final SpecConfigFulu specConfigFulu = SpecConfigFulu.required(config); + return getCurrentEpoch(store) + .minusMinZero(epoch) + .isLessThanOrEqualTo(specConfigFulu.getMinEpochsForDataColumnSidecarsRequests()); + } + public Optional computeFirstSlotWithBlobSupport() { return getSpecConfigDeneb() .map(SpecConfigDeneb::getDenebForkEpoch) @@ -1020,6 +1042,15 @@ private Optional getSpecConfigDeneb() { .flatMap(SpecConfig::toVersionDeneb); } + // Fulu private helpers + private Optional getSpecConfigFulu() { + final SpecMilestone highestSupportedMilestone = + getForkSchedule().getHighestSupportedMilestone(); + return Optional.ofNullable(forMilestone(highestSupportedMilestone)) + .map(SpecVersion::getConfig) + .flatMap(SpecConfig::toVersionFulu); + } + // Private helpers private SpecVersion atState(final BeaconState state) { return atSlot(state.getSlot()); diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/execution/BlobAndCellProofs.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/execution/BlobAndCellProofs.java new file mode 100644 index 00000000000..6515242721d --- /dev/null +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/execution/BlobAndCellProofs.java @@ -0,0 +1,33 @@ +/* + * 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.spec.datastructures.execution; + +import static com.google.common.base.Preconditions.checkArgument; +import static tech.pegasys.teku.kzg.KZG.CELLS_PER_EXT_BLOB; + +import java.util.List; +import tech.pegasys.teku.kzg.KZGProof; +import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.Blob; + +public record BlobAndCellProofs(Blob blob, List cellProofs) { + public BlobAndCellProofs(final Blob blob, final List cellProofs) { + checkArgument( + cellProofs.size() == CELLS_PER_EXT_BLOB, + "Expected %s proofs but got %s", + CELLS_PER_EXT_BLOB, + cellProofs.size()); + this.blob = blob; + this.cellProofs = cellProofs; + } +} diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MathHelpers.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MathHelpers.java index ec0811e52f0..739069bf50b 100644 --- a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MathHelpers.java +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MathHelpers.java @@ -18,6 +18,7 @@ import java.nio.ByteOrder; import org.apache.tuweni.bytes.Bytes; import org.apache.tuweni.bytes.Bytes32; +import org.apache.tuweni.units.bigints.UInt256; import tech.pegasys.teku.infrastructure.unsigned.UInt64; public class MathHelpers { @@ -134,4 +135,20 @@ static Bytes32 uintToBytes32(final UInt64 value) { public static UInt64 bytesToUInt64(final Bytes data) { return UInt64.fromLongBits(data.toLong(ByteOrder.LITTLE_ENDIAN)); } + + public static Bytes uint256ToBytes(final UInt256 number) { + final Bytes intBytes = + Bytes.wrap(number.toUnsignedBigInteger(ByteOrder.LITTLE_ENDIAN).toByteArray()) + .trimLeadingZeros(); + // We should keep 32 bytes + return Bytes32.leftPad(intBytes); + } + + public static int intPlusMaxIntCapped(final int a, final int b) { + final UInt64 sum = UInt64.valueOf(a).plus(b); + if (sum.isLessThanOrEqualTo(UInt64.valueOf(Integer.MAX_VALUE))) { + return sum.intValue(); + } + return Integer.MAX_VALUE; + } } diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MiscHelpers.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MiscHelpers.java index 45b252d7b00..050f7cb2009 100644 --- a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MiscHelpers.java +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/common/helpers/MiscHelpers.java @@ -217,7 +217,7 @@ public List computeSubscribedSubnets(final UInt256 nodeId, final UInt64 .toList(); } - private UInt64 computeSubscribedSubnet( + protected UInt64 computeSubscribedSubnet( final UInt256 nodeId, final UInt64 epoch, final int index) { final int nodeIdPrefix = diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDeneb.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDeneb.java index 15621b2cc12..ea90a4cac24 100644 --- a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDeneb.java +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDeneb.java @@ -55,6 +55,7 @@ public class MiscHelpersDeneb extends MiscHelpersCapella { private final Predicates predicates; private final BeaconBlockBodySchemaDeneb beaconBlockBodySchema; private final BlobSidecarSchema blobSidecarSchema; + private final SpecConfigDeneb specConfigDeneb; public static MiscHelpersDeneb required(final MiscHelpers miscHelpers) { return miscHelpers @@ -71,6 +72,7 @@ public MiscHelpersDeneb( final Predicates predicates, final SchemaDefinitionsDeneb schemaDefinitions) { super(specConfig); + this.specConfigDeneb = specConfig; this.predicates = predicates; this.beaconBlockBodySchema = (BeaconBlockBodySchemaDeneb) schemaDefinitions.getBeaconBlockBodySchema(); @@ -240,7 +242,7 @@ public int getBlobSidecarKzgCommitmentGeneralizedIndex(final UInt64 blobSidecarI GIndexUtil.gIdxCompose(blobKzgCommitmentsGeneralizedIndex, commitmentGeneralizedIndex); } - public List computeKzgCommitmentInclusionProof( + public List computeBlobKzgCommitmentInclusionProof( final UInt64 blobSidecarIndex, final BeaconBlockBody beaconBlockBody) { return MerkleUtil.constructMerkleProof( beaconBlockBody.getBackingNode(), @@ -265,7 +267,7 @@ public BlobSidecar constructBlobSidecar( index, commitmentsCount)); } final List kzgCommitmentInclusionProof = - computeKzgCommitmentInclusionProof(index, beaconBlockBody); + computeBlobKzgCommitmentInclusionProof(index, beaconBlockBody); return blobSidecarSchema.create( index, blob, commitment, proof, signedBeaconBlock.asHeader(), kzgCommitmentInclusionProof); } @@ -286,7 +288,8 @@ public BlobSidecar constructBlobSidecarFromBlobAndProof( sszKZGCommitment, new SszKZGProof(blobAndProof.proof()), signedBeaconBlockHeader, - computeKzgCommitmentInclusionProof(blobIdentifier.getIndex(), beaconBlockBodyDeneb)); + computeBlobKzgCommitmentInclusionProof( + blobIdentifier.getIndex(), beaconBlockBodyDeneb)); blobSidecar.markSignatureAsValidated(); blobSidecar.markKzgCommitmentInclusionProofAsValidated(); @@ -314,4 +317,11 @@ public boolean verifyBlobKzgCommitmentInclusionProof(final BlobSidecar blobSidec return result; } + + public boolean isAvailabilityOfBlobSidecarsRequiredAtEpoch( + final UInt64 currentEpoch, final UInt64 epoch) { + return currentEpoch + .minusMinZero(epoch) + .isLessThanOrEqualTo(specConfigDeneb.getMinEpochsForBlobSidecarsRequests()); + } } diff --git a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/fulu/helpers/MiscHelpersFulu.java b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/fulu/helpers/MiscHelpersFulu.java index f9b2a76b7f7..275df2d61e7 100644 --- a/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/fulu/helpers/MiscHelpersFulu.java +++ b/ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/fulu/helpers/MiscHelpersFulu.java @@ -13,9 +13,54 @@ package tech.pegasys.teku.spec.logic.versions.fulu.helpers; +import static tech.pegasys.teku.spec.logic.common.helpers.MathHelpers.bytesToUInt64; +import static tech.pegasys.teku.spec.logic.common.helpers.MathHelpers.uint256ToBytes; + +import com.google.common.annotations.VisibleForTesting; +import java.math.BigDecimal; +import java.math.MathContext; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import java.util.stream.Stream; +import org.apache.tuweni.bytes.Bytes32; +import org.apache.tuweni.units.bigints.UInt256; +import tech.pegasys.teku.infrastructure.crypto.Hash; +import tech.pegasys.teku.infrastructure.ssz.SszList; +import tech.pegasys.teku.infrastructure.ssz.schema.SszListSchema; +import tech.pegasys.teku.infrastructure.ssz.tree.MerkleUtil; +import tech.pegasys.teku.infrastructure.unsigned.UInt64; +import tech.pegasys.teku.kzg.KZG; +import tech.pegasys.teku.kzg.KZGCell; +import tech.pegasys.teku.kzg.KZGCellAndProof; +import tech.pegasys.teku.kzg.KZGCellID; +import tech.pegasys.teku.kzg.KZGCellWithColumnId; import tech.pegasys.teku.spec.config.SpecConfigElectra; import tech.pegasys.teku.spec.config.SpecConfigFulu; +import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.Blob; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.Cell; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumn; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSchema; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSidecar; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSidecarSchema; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.MatrixEntry; +import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlock; +import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock; +import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlockHeader; +import tech.pegasys.teku.spec.datastructures.blocks.blockbody.BeaconBlockBody; +import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.deneb.BeaconBlockBodyDeneb; +import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.deneb.BlindedBeaconBlockBodyDeneb; +import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.electra.BeaconBlockBodySchemaElectra; +import tech.pegasys.teku.spec.datastructures.execution.BlobAndCellProofs; +import tech.pegasys.teku.spec.datastructures.state.Validator; +import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState; +import tech.pegasys.teku.spec.datastructures.type.SszKZGCommitment; +import tech.pegasys.teku.spec.datastructures.type.SszKZGProof; import tech.pegasys.teku.spec.logic.common.helpers.MiscHelpers; import tech.pegasys.teku.spec.logic.common.helpers.Predicates; import tech.pegasys.teku.spec.logic.versions.electra.helpers.MiscHelpersElectra; @@ -24,6 +69,7 @@ import tech.pegasys.teku.spec.schemas.SchemaDefinitionsFulu; public class MiscHelpersFulu extends MiscHelpersElectra { + private static final MathContext BIGDECIMAL_PRECISION = MathContext.DECIMAL128; public static MiscHelpersFulu required(final MiscHelpers miscHelpers) { return miscHelpers @@ -61,4 +107,471 @@ public MiscHelpersFulu( public Optional toVersionFulu() { return Optional.of(this); } + + private UInt256 incrementByModule(final UInt256 n) { + if (n.equals(UInt256.MAX_VALUE)) { + return UInt256.ZERO; + } else { + return n.plus(1); + } + } + + public UInt64 computeSubnetForDataColumnSidecar(final UInt64 columnIndex) { + return columnIndex.mod(specConfigFulu.getDataColumnSidecarSubnetCount()); + } + + public List computeDataColumnSidecarBackboneSubnets( + final UInt256 nodeId, final UInt64 epoch, final int groupCount) { + final List columns = computeCustodyColumnIndexes(nodeId, groupCount); + return columns.stream().map(this::computeSubnetForDataColumnSidecar).toList(); + } + + public List computeCustodyColumnIndexes(final UInt256 nodeId, final int groupCount) { + final List custodyGroups = getCustodyGroups(nodeId, groupCount); + return custodyGroups.stream() + .flatMap(group -> computeColumnsForCustodyGroup(group).stream()) + .toList(); + } + + public List computeColumnsForCustodyGroup(final UInt64 custodyGroup) { + if (custodyGroup.isGreaterThanOrEqualTo(specConfigFulu.getNumberOfCustodyGroups())) { + throw new IllegalArgumentException( + String.format( + "Custody group %s couldn't exceed number of groups %s", + custodyGroup, specConfigFulu.getNumberOfCustodyGroups())); + } + + final int columnsPerGroup = + specConfigFulu.getNumberOfColumns() / specConfigFulu.getNumberOfCustodyGroups(); + + return IntStream.range(0, columnsPerGroup) + .mapToLong( + i -> (long) specConfigFulu.getNumberOfCustodyGroups() * i + custodyGroup.intValue()) + .sorted() + .mapToObj(UInt64::valueOf) + .toList(); + } + + private UInt64 computeCustodyGroupIndex(final UInt256 nodeId) { + return bytesToUInt64(Hash.sha256(uint256ToBytes(nodeId)).slice(0, 8)) + .mod(specConfigFulu.getNumberOfCustodyGroups()); + } + + public List getCustodyGroups(final UInt256 nodeId, final int custodyGroupCount) { + if (custodyGroupCount > specConfigFulu.getNumberOfCustodyGroups()) { + throw new IllegalArgumentException( + String.format( + "Custody group count %s couldn't exceed number of groups %s", + custodyGroupCount, specConfigFulu.getNumberOfCustodyGroups())); + } + + return Stream.iterate(nodeId, this::incrementByModule) + .map(this::computeCustodyGroupIndex) + .distinct() + .limit(custodyGroupCount) + .sorted() + .toList(); + } + + public UInt64 getValidatorsCustodyRequirement( + final BeaconState state, final Set validatorIndices) { + final UInt64 totalNodeBalance = + validatorIndices.stream() + .map( + proposerIndex -> { + final Validator validator = state.getValidators().get(proposerIndex.intValue()); + return validator.getEffectiveBalance(); + }) + .reduce(UInt64.ZERO, UInt64::plus); + final UInt64 count = + totalNodeBalance.dividedBy(specConfigFulu.getBalancePerAdditionalCustodyGroup()); + return count + .max(specConfigFulu.getValidatorCustodyRequirement()) + .min(specConfigFulu.getNumberOfCustodyGroups()); + } + + public boolean verifyDataColumnSidecarKzgProof( + final KZG kzg, final DataColumnSidecar dataColumnSidecar) { + final int dataColumns = specConfigFulu.getNumberOfColumns(); + if (dataColumnSidecar.getIndex().isGreaterThanOrEqualTo(dataColumns)) { + return false; + } + + // Number of rows is the same for cells, commitments, proofs + if (dataColumnSidecar.getDataColumn().size() != dataColumnSidecar.getSszKZGCommitments().size() + || dataColumnSidecar.getSszKZGCommitments().size() + != dataColumnSidecar.getSszKZGProofs().size()) { + return false; + } + + final List cellWithIds = + IntStream.range(0, dataColumnSidecar.getDataColumn().size()) + .mapToObj( + rowIndex -> + KZGCellWithColumnId.fromCellAndColumn( + new KZGCell(dataColumnSidecar.getDataColumn().get(rowIndex).getBytes()), + dataColumnSidecar.getIndex().intValue())) + .collect(Collectors.toList()); + + return kzg.verifyCellProofBatch( + dataColumnSidecar.getSszKZGCommitments().stream() + .map(SszKZGCommitment::getKZGCommitment) + .toList(), + cellWithIds, + dataColumnSidecar.getSszKZGProofs().stream().map(SszKZGProof::getKZGProof).toList()); + } + + public boolean verifyDataColumnSidecarInclusionProof(final DataColumnSidecar dataColumnSidecar) { + if (dataColumnSidecar.getSszKZGCommitments().isEmpty()) { + return false; + } + return predicates.isValidMerkleBranch( + dataColumnSidecar.getSszKZGCommitments().hashTreeRoot(), + dataColumnSidecar.getKzgCommitmentsInclusionProof(), + specConfigFulu.getKzgCommitmentsInclusionProofDepth().intValue(), + getBlockBodyKzgCommitmentsGeneralizedIndex(), + dataColumnSidecar.getBlockBodyRoot()); + } + + public int getBlockBodyKzgCommitmentsGeneralizedIndex() { + return (int) + BeaconBlockBodySchemaElectra.required(schemaDefinitions.getBeaconBlockBodySchema()) + .getBlobKzgCommitmentsGeneralizedIndex(); + } + + public List computeDataColumnKzgCommitmentsInclusionProof( + final BeaconBlockBody beaconBlockBody) { + return MerkleUtil.constructMerkleProof( + beaconBlockBody.getBackingNode(), getBlockBodyKzgCommitmentsGeneralizedIndex()); + } + + @VisibleForTesting + @Deprecated + public List constructDataColumnSidecarsOld( + final SignedBeaconBlock signedBeaconBlock, final List blobs, final KZG kzg) { + return constructDataColumnSidecars( + signedBeaconBlock.getMessage(), + signedBeaconBlock.asHeader(), + computeExtendedMatrixAndProofs(blobs, kzg)); + } + + public List constructDataColumnSidecars( + final SignedBeaconBlock signedBeaconBlock, + final List blobAndCellProofsList, + final KZG kzg) { + return constructDataColumnSidecars( + signedBeaconBlock.getMessage(), + signedBeaconBlock.asHeader(), + computeExtendedMatrix(blobAndCellProofsList, kzg)); + } + + public List constructDataColumnSidecars( + final SignedBeaconBlockHeader signedBeaconBlockHeader, + final SszList sszKZGCommitments, + final List kzgCommitmentsInclusionProof, + final List blobAndCellProofsList, + final KZG kzg) { + final List> extendedMatrix = + computeExtendedMatrix(blobAndCellProofsList, kzg); + return constructDataColumnSidecars( + signedBeaconBlockHeader, sszKZGCommitments, kzgCommitmentsInclusionProof, extendedMatrix); + } + + /** + * Return the full ``ExtendedMatrix``. + * + *

This helper demonstrates the relationship between blobs and ``ExtendedMatrix``. + * + *

>The data structure for storing cells is implementation-dependent. + */ + public List> computeExtendedMatrixAndProofs( + final List blobs, final KZG kzg) { + return IntStream.range(0, blobs.size()) + .parallel() + .mapToObj( + blobIndex -> { + final List kzgCellAndProofs = + kzg.computeCellsAndProofs(blobs.get(blobIndex).getBytes()); + final List row = new ArrayList<>(); + for (int cellIndex = 0; cellIndex < kzgCellAndProofs.size(); ++cellIndex) { + row.add( + schemaDefinitions + .getMatrixEntrySchema() + .create( + kzgCellAndProofs.get(cellIndex).cell(), + kzgCellAndProofs.get(cellIndex).proof(), + blobIndex, + cellIndex)); + } + return row; + }) + .toList(); + } + + public List> computeExtendedMatrix( + final List blobAndCellProofsList, final KZG kzg) { + return IntStream.range(0, blobAndCellProofsList.size()) + .parallel() + .mapToObj( + blobIndex -> { + final BlobAndCellProofs blobAndCellProofs = blobAndCellProofsList.get(blobIndex); + final List kzgCells = kzg.computeCells(blobAndCellProofs.blob().getBytes()); + final List row = new ArrayList<>(); + for (int cellIndex = 0; cellIndex < kzgCells.size(); ++cellIndex) { + row.add( + schemaDefinitions + .getMatrixEntrySchema() + .create( + kzgCells.get(cellIndex), + blobAndCellProofs.cellProofs().get(cellIndex), + blobIndex, + cellIndex)); + } + return row; + }) + .toList(); + } + + @VisibleForTesting + public List constructDataColumnSidecars( + final BeaconBlock beaconBlock, + final SignedBeaconBlockHeader signedBeaconBlockHeader, + final List> extendedMatrix) { + if (extendedMatrix.isEmpty()) { + return Collections.emptyList(); + } + + final SszList sszKZGCommitments; + final List kzgCommitmentsInclusionProof; + if (beaconBlock.isBlinded()) { + final BlindedBeaconBlockBodyDeneb beaconBlockBody = + BlindedBeaconBlockBodyDeneb.required(beaconBlock.getBody()); + sszKZGCommitments = beaconBlockBody.getBlobKzgCommitments(); + kzgCommitmentsInclusionProof = computeDataColumnKzgCommitmentsInclusionProof(beaconBlockBody); + } else { + final BeaconBlockBodyDeneb beaconBlockBody = + BeaconBlockBodyDeneb.required(beaconBlock.getBody()); + sszKZGCommitments = beaconBlockBody.getBlobKzgCommitments(); + kzgCommitmentsInclusionProof = computeDataColumnKzgCommitmentsInclusionProof(beaconBlockBody); + } + + return constructDataColumnSidecars( + signedBeaconBlockHeader, sszKZGCommitments, kzgCommitmentsInclusionProof, extendedMatrix); + } + + private List constructDataColumnSidecars( + final SignedBeaconBlockHeader signedBeaconBlockHeader, + final SszList sszKZGCommitments, + final List kzgCommitmentsInclusionProof, + final List> extendedMatrix) { + if (extendedMatrix.isEmpty()) { + return Collections.emptyList(); + } + + final DataColumnSchema dataColumnSchema = schemaDefinitions.getDataColumnSchema(); + final DataColumnSidecarSchema dataColumnSidecarSchema = + schemaDefinitions.getDataColumnSidecarSchema(); + final SszListSchema kzgProofsSchema = + dataColumnSidecarSchema.getKzgProofsSchema(); + + final int columnCount = extendedMatrix.getFirst().size(); + + return IntStream.range(0, columnCount) + .mapToObj( + cellID -> { + List columnData = + extendedMatrix.stream().map(row -> row.get(cellID)).toList(); + List columnCells = columnData.stream().map(MatrixEntry::getCell).toList(); + + SszList columnProofs = + columnData.stream() + .map(MatrixEntry::getKzgProof) + .map(SszKZGProof::new) + .collect(kzgProofsSchema.collector()); + final DataColumn dataColumn = dataColumnSchema.create(columnCells); + + return dataColumnSidecarSchema.create( + UInt64.valueOf(cellID), + dataColumn, + sszKZGCommitments, + columnProofs, + signedBeaconBlockHeader, + kzgCommitmentsInclusionProof); + }) + .toList(); + } + + public List reconstructAllDataColumnSidecars( + final Collection existingSidecars, final KZG kzg) { + if (existingSidecars.size() < (specConfigFulu.getNumberOfColumns() / 2)) { + throw new IllegalArgumentException( + "Number of sidecars must be greater than or equal to the half of column count"); + } + final List> columnBlobEntries = + existingSidecars.stream() + .map( + sideCar -> + IntStream.range(0, sideCar.getDataColumn().size()) + .mapToObj( + rowIndex -> + schemaDefinitions + .getMatrixEntrySchema() + .create( + sideCar.getDataColumn().get(rowIndex), + sideCar.getSszKZGProofs().get(rowIndex).getKZGProof(), + sideCar.getIndex(), + UInt64.valueOf(rowIndex))) + .toList()) + .toList(); + final List> blobColumnEntries = transpose(columnBlobEntries); + final List> extendedMatrix = recoverMatrix(blobColumnEntries, kzg); + final DataColumnSidecar anyExistingSidecar = + existingSidecars.stream().findFirst().orElseThrow(); + final SignedBeaconBlockHeader signedBeaconBlockHeader = + anyExistingSidecar.getSignedBeaconBlockHeader(); + return constructDataColumnSidecars( + signedBeaconBlockHeader, + anyExistingSidecar.getSszKZGCommitments(), + anyExistingSidecar.getKzgCommitmentsInclusionProof().asListUnboxed(), + extendedMatrix); + } + + /** + * Return the recovered extended matrix. + * + *

This helper demonstrates how to apply ``recover_cells_and_kzg_proofs``. + * + *

The data structure for storing cells is implementation-dependent. + */ + public List> recoverMatrix( + final List> partialMatrix, final KZG kzg) { + return IntStream.range(0, partialMatrix.size()) + .parallel() + .mapToObj( + blobIndex -> { + final List cellWithColumnIds = + partialMatrix.get(blobIndex).stream() + .filter(entry -> entry.getRowIndex().intValue() == blobIndex) + .map( + entry -> + new KZGCellWithColumnId( + new KZGCell(entry.getCell().getBytes()), + new KZGCellID(entry.getColumnIndex()))) + .toList(); + final List kzgCellAndProofs = + kzg.recoverCellsAndProofs(cellWithColumnIds); + return IntStream.range(0, kzgCellAndProofs.size()) + .mapToObj( + kzgCellAndProofIndex -> + schemaDefinitions + .getMatrixEntrySchema() + .create( + kzgCellAndProofs.get(kzgCellAndProofIndex).cell(), + kzgCellAndProofs.get(kzgCellAndProofIndex).proof(), + kzgCellAndProofIndex, + blobIndex)) + .toList(); + }) + .toList(); + } + + /** + * Return the sample count if allowing failures. + * + *

This helper demonstrates how to calculate the number of columns to query per slot when + * allowing given number of failures, assuming uniform random selection without replacement. + * Nested functions are direct replacements of Python library functions math.comb and + * scipy.stats.hypergeom.cdf, with the same signatures. + */ + public UInt64 getExtendedSampleCount(final UInt64 allowedFailures) { + if (allowedFailures.isGreaterThan(specConfigFulu.getNumberOfColumns() / 2)) { + throw new IllegalArgumentException( + String.format( + "Allowed failures (%s) should be less than half of columns number (%s)", + allowedFailures, specConfigFulu.getNumberOfColumns())); + } + final UInt64 worstCaseMissing = UInt64.valueOf(specConfigFulu.getNumberOfColumns() / 2 + 1); + final double falsePositiveThreshold = + hypergeomCdf( + UInt64.ZERO, + UInt64.valueOf(specConfigFulu.getNumberOfColumns()), + worstCaseMissing, + UInt64.valueOf(specConfigFulu.getSamplesPerSlot())); + UInt64 sampleCount = UInt64.valueOf(specConfigFulu.getSamplesPerSlot()); + for (; + sampleCount.isLessThanOrEqualTo(specConfigFulu.getNumberOfColumns()); + sampleCount = sampleCount.increment()) { + if (hypergeomCdf( + allowedFailures, + UInt64.valueOf(specConfigFulu.getNumberOfColumns()), + worstCaseMissing, + sampleCount) + <= falsePositiveThreshold) { + break; + } + } + return sampleCount; + } + + private static List> transpose(final List> matrix) { + final int rowCount = matrix.size(); + final int colCount = matrix.getFirst().size(); + final List> ret = + Stream.generate(() -> (List) new ArrayList(rowCount)).limit(colCount).toList(); + + for (int row = 0; row < rowCount; row++) { + if (matrix.get(row).size() != colCount) { + throw new IllegalArgumentException("Different number columns in the matrix"); + } + for (int col = 0; col < colCount; col++) { + final T val = matrix.get(row).get(col); + ret.get(col).add(row, val); + } + } + return ret; + } + + private UInt256 mathComb(final UInt64 n, final UInt64 k) { + if (n.isGreaterThanOrEqualTo(k)) { + UInt256 r = UInt256.ONE; + for (UInt64 i = UInt64.ZERO; + i.isLessThan(k.isGreaterThan(n.minus(k)) ? n.minus(k) : k); + i = i.plus(1)) { + r = r.multiply(n.minus(i).longValue()).divide(i.plus(1).longValue()); + } + return r; + } else { + return UInt256.ZERO; + } + } + + @SuppressWarnings("JavaCase") + private double hypergeomCdf(final UInt64 k, final UInt64 M, final UInt64 n, final UInt64 N) { + return Stream.iterate(UInt64.ZERO, i -> i.isLessThanOrEqualTo(k), UInt64::increment) + .mapToDouble( + i -> + N.isLessThan(i) + ? 0d + : new BigDecimal( + mathComb(n, i) + .multiply(mathComb(M.minus(n), N.minus(i))) + .toBigInteger(), + BIGDECIMAL_PRECISION) + .divide(new BigDecimal(mathComb(M, N).toBigInteger()), BIGDECIMAL_PRECISION) + .doubleValue()) + .sum(); + } + + @Override + public boolean isAvailabilityOfBlobSidecarsRequiredAtEpoch( + final UInt64 currentEpoch, final UInt64 epoch) { + return !epoch.isGreaterThanOrEqualTo(specConfigFulu.getFuluForkEpoch()); + } + + public boolean isAvailabilityOfDataColumnSidecarsRequiredAtEpoch( + final UInt64 currentEpoch, final UInt64 epoch) { + return currentEpoch + .minusMinZero(epoch) + .isLessThanOrEqualTo(specConfigFulu.getMinEpochsForDataColumnSidecarsRequests()); + } } diff --git a/ethereum/spec/src/test/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDenebTest.java b/ethereum/spec/src/test/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDenebTest.java index dadfb309a0b..eeedecf7396 100644 --- a/ethereum/spec/src/test/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDenebTest.java +++ b/ethereum/spec/src/test/java/tech/pegasys/teku/spec/logic/versions/deneb/helpers/MiscHelpersDenebTest.java @@ -163,7 +163,8 @@ void verifyBlobKzgCommitmentInclusionProofShouldValidate() { for (int i = 0; i < numberOfCommitments; ++i) { final UInt64 blobSidecarIndex = UInt64.valueOf(i); final List merkleProof = - miscHelpersDeneb.computeKzgCommitmentInclusionProof(blobSidecarIndex, beaconBlockBody); + miscHelpersDeneb.computeBlobKzgCommitmentInclusionProof( + blobSidecarIndex, beaconBlockBody); assertThat(merkleProof.size()) .isEqualTo( SpecConfigDeneb.required(spec.getGenesisSpecConfig()) @@ -193,7 +194,7 @@ void verifyBlobKzgCommitmentInclusionProofShouldValidate() { final UInt64 wrongIndex = UInt64.valueOf(j); final List merkleProofWrong = - miscHelpersDeneb.computeKzgCommitmentInclusionProof(wrongIndex, beaconBlockBody); + miscHelpersDeneb.computeBlobKzgCommitmentInclusionProof(wrongIndex, beaconBlockBody); assertThat(merkleProofWrong.size()) .isEqualTo( SpecConfigDeneb.required(spec.getGenesisSpecConfig()) diff --git a/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/generator/ChainBuilder.java b/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/generator/ChainBuilder.java index bc3eeabdac2..ddd7272b5c6 100644 --- a/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/generator/ChainBuilder.java +++ b/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/generator/ChainBuilder.java @@ -737,7 +737,7 @@ private SignedBlockAndState generateBlockWithBlobSidecars( final Blob blob = blobs.get(index); final KZGCommitment kzgCommitment = kzgCommitments.get(index); final List merkleProof = - miscHelpersDeneb.computeKzgCommitmentInclusionProof( + miscHelpersDeneb.computeBlobKzgCommitmentInclusionProof( blobSidecarIndex, nextBlockAndState.getBlock().getMessage().getBody()); return new BlobSidecar( blobSidecarSchema, diff --git a/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/util/DataStructureUtil.java b/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/util/DataStructureUtil.java index 778b7d5bdac..17ceff1b255 100644 --- a/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/util/DataStructureUtil.java +++ b/ethereum/spec/src/testFixtures/java/tech/pegasys/teku/spec/util/DataStructureUtil.java @@ -2560,7 +2560,7 @@ public List randomKzgCommitmentInclusionProof() { public List validKzgCommitmentInclusionProof( final UInt64 blobIndex, final BeaconBlockBody beaconBlockBody) { return MiscHelpersDeneb.required(spec.forMilestone(SpecMilestone.DENEB).miscHelpers()) - .computeKzgCommitmentInclusionProof(blobIndex, beaconBlockBody); + .computeBlobKzgCommitmentInclusionProof(blobIndex, beaconBlockBody); } public SszList randomBlobKzgCommitments() { diff --git a/ethereum/statetransition/build.gradle b/ethereum/statetransition/build.gradle index 5016052cf6a..dd987ab74a1 100644 --- a/ethereum/statetransition/build.gradle +++ b/ethereum/statetransition/build.gradle @@ -26,6 +26,7 @@ dependencies { implementation project(':storage:api') implementation 'io.consensys.tuweni:tuweni-units' + implementation 'io.libp2p:jvm-libp2p' testImplementation testFixtures(project(':ethereum:spec')) testImplementation testFixtures(project(':ethereum:networks')) diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/RemoteOrigin.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/RemoteOrigin.java new file mode 100644 index 00000000000..f76d1feed6c --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/RemoteOrigin.java @@ -0,0 +1,22 @@ +/* + * 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.blobs; + +public enum RemoteOrigin { + RPC, + GOSSIP, + LOCAL_EL, + LOCAL_PROPOSAL, + RECOVERED +} diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipBatchLogger.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipBatchLogger.java new file mode 100644 index 00000000000..7dedff4c32c --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipBatchLogger.java @@ -0,0 +1,254 @@ +/* + * 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.statetransition.datacolumns.log.gossip; + +import com.google.common.base.Throwables; +import io.libp2p.pubsub.MessageAlreadySeenException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.stream.Collectors; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import tech.pegasys.teku.infrastructure.async.AsyncRunner; +import tech.pegasys.teku.infrastructure.logging.LogFormatter; +import tech.pegasys.teku.infrastructure.time.TimeProvider; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSidecar; +import tech.pegasys.teku.spec.datastructures.blocks.SlotAndBlockRoot; +import tech.pegasys.teku.statetransition.datacolumns.util.StringifyUtil; +import tech.pegasys.teku.statetransition.validation.InternalValidationResult; +import tech.pegasys.teku.statetransition.validation.ValidationResultCode; + +public class DasGossipBatchLogger implements DasGossipLogger { + private static final Logger LOG = LogManager.getLogger(DasGossipLogger.class); + private final TimeProvider timeProvider; + + private List events = new ArrayList<>(); + + public DasGossipBatchLogger(final AsyncRunner asyncRunner, final TimeProvider timeProvider) { + this.timeProvider = timeProvider; + asyncRunner.runWithFixedDelay( + this::logBatchedEvents, + Duration.ofSeconds(1), + err -> LOG.info("DasGossipBatchLogger error: {}", err.toString())); + } + + interface Event { + + long time(); + } + + interface ColumnEvent extends Event { + DataColumnSidecar sidecar(); + } + + record ReceiveEvent( + long time, DataColumnSidecar sidecar, InternalValidationResult validationResult) + implements ColumnEvent {} + + record PublishEvent(long time, DataColumnSidecar sidecar, Optional result) + implements ColumnEvent {} + + record SubscribeEvent(long time, int subnetId) implements Event {} + + record UnsubscribeEvent(long time, int subnetId) implements Event {} + + private void logBatchedEvents() { + final List eventsLoc; + synchronized (this) { + if (events.isEmpty()) { + return; + } + eventsLoc = events; + events = new ArrayList<>(); + } + + groupByBlock(ReceiveEvent.class, eventsLoc).forEach(this::logReceiveEvents); + groupByBlock(PublishEvent.class, eventsLoc).forEach(this::logPublishEvents); + logSubscriptionEvents(eventsLoc); + } + + private void logReceiveEvents(final SlotAndBlockRoot blockId, final List events) { + final Map> eventsByValidateCode = + events.stream().collect(Collectors.groupingBy(e -> e.validationResult().code())); + eventsByValidateCode.forEach( + (validationCode, codeEvents) -> { + Level level = validationCode == ValidationResultCode.REJECT ? Level.INFO : Level.DEBUG; + LOG.log( + level, + "Received {} data columns (validation result: {}) by gossip {} for block {}: {}", + codeEvents.size(), + validationCode, + msAgoString(codeEvents), + blockIdString(blockId), + columnIndexesString(codeEvents)); + }); + } + + private void logPublishEvents(final SlotAndBlockRoot blockId, final List events) { + final Map>, List> eventsByError = + events.stream() + .collect( + Collectors.groupingBy( + e -> e.result().map(thr -> ExceptionUtils.getRootCause(thr).getClass()))); + eventsByError.forEach( + (maybeErrorClass, errEvents) -> { + Optional someError = errEvents.getFirst().result(); + someError.ifPresentOrElse( + error -> logErrorByType(error, events, blockId), + () -> { + LOG.debug( + "Published {} data columns by gossip {} for block {}: {}", + errEvents.size(), + msAgoString(errEvents), + blockIdString(blockId), + columnIndexesString(errEvents)); + }); + }); + } + + private void logErrorByType( + final Throwable error, final List errEvents, final SlotAndBlockRoot blockId) { + final Throwable rootCause = Throwables.getRootCause(error); + switch (rootCause) { + case MessageAlreadySeenException ignored -> + LOG.debug( + "Error publishing {} data columns ({}) by gossip {} for block {}: has already been seen", + errEvents.size(), + columnIndexesString(errEvents), + msAgoString(errEvents), + blockIdString(blockId)); + default -> + LOG.info( + "Error publishing {} data columns ({}) by gossip {} for block {}: {}", + errEvents.size(), + columnIndexesString(errEvents), + msAgoString(errEvents), + blockIdString(blockId), + error); + } + } + + private void logSubscriptionEvents(final List events) { + final List subscribedSubnets = new ArrayList<>(); + final List unsubscribedSubnets = new ArrayList<>(); + events.forEach( + e -> { + switch (e) { + case SubscribeEvent event -> subscribedSubnets.add(event.subnetId()); + case UnsubscribeEvent event -> unsubscribedSubnets.add(event.subnetId()); + default -> {} + } + }); + + if (!(subscribedSubnets.isEmpty() && unsubscribedSubnets.isEmpty())) { + String subscribeString = + subscribedSubnets.isEmpty() + ? "" + : "subscribed: " + StringifyUtil.toIntRangeStringWithSize(subscribedSubnets); + String unsubscribeString = + unsubscribedSubnets.isEmpty() + ? "" + : "unsubscribed: " + StringifyUtil.toIntRangeStringWithSize(unsubscribedSubnets); + String maybeDelim = subscribedSubnets.isEmpty() || unsubscribedSubnets.isEmpty() ? "" : ", "; + LOG.info( + "Data column gossip subnets subscriptions changed: " + + subscribeString + + maybeDelim + + unsubscribeString); + } + } + + private String columnIndexesString(final List events) { + final List columnIndexes = + events.stream().map(e -> e.sidecar().getIndex().intValue()).toList(); + return StringifyUtil.toIntRangeString(columnIndexes); + } + + private static String blockIdString(final SlotAndBlockRoot blockId) { + return "#" + + blockId.getSlot() + + " (0x" + + LogFormatter.formatAbbreviatedHashRoot(blockId.getBlockRoot()) + + ")"; + } + + private String msAgoString(final List events) { + long curTime = timeProvider.getTimeInMillis().longValue(); + long firstMillisAgo = curTime - events.getFirst().time(); + long lastMillisAgo = curTime - events.getLast().time(); + return (lastMillisAgo == firstMillisAgo + ? lastMillisAgo + "ms" + : lastMillisAgo + "ms-" + firstMillisAgo + "ms") + + " ago"; + } + + private boolean needToLogEvent(final boolean isSevereEvent) { + return LOG.isDebugEnabled() || (isSevereEvent && LOG.isInfoEnabled()); + } + + @Override + public synchronized void onReceive( + final DataColumnSidecar sidecar, final InternalValidationResult validationResult) { + if (needToLogEvent(validationResult.isReject())) { + events.add( + new ReceiveEvent(timeProvider.getTimeInMillis().longValue(), sidecar, validationResult)); + } + } + + @Override + public synchronized void onPublish( + final DataColumnSidecar sidecar, final Optional result) { + if (needToLogEvent(result.isPresent())) { + events.add(new PublishEvent(timeProvider.getTimeInMillis().longValue(), sidecar, result)); + } + } + + @Override + public void onDataColumnSubnetSubscribe(final int subnetId) { + if (needToLogEvent(false)) { + events.add(new SubscribeEvent(timeProvider.getTimeInMillis().longValue(), subnetId)); + } + } + + @Override + public void onDataColumnSubnetUnsubscribe(final int subnetId) { + if (needToLogEvent(false)) { + events.add(new UnsubscribeEvent(timeProvider.getTimeInMillis().longValue(), subnetId)); + } + } + + private static + SortedMap> groupByBlock( + final Class eventClass, final List allEvents) { + final SortedMap> eventsByBlock = new TreeMap<>(); + for (final Event event : allEvents) { + if (eventClass.isAssignableFrom(event.getClass())) { + @SuppressWarnings("unchecked") + final TEvent e = (TEvent) event; + final DataColumnSidecar sidecar = e.sidecar(); + final SlotAndBlockRoot blockId = + new SlotAndBlockRoot(sidecar.getSlot(), sidecar.getBlockRoot()); + eventsByBlock.computeIfAbsent(blockId, __ -> new ArrayList<>()).add(e); + } + } + return eventsByBlock; + } +} diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipLogger.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipLogger.java new file mode 100644 index 00000000000..3069836456c --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/DasGossipLogger.java @@ -0,0 +1,37 @@ +/* + * 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.statetransition.datacolumns.log.gossip; + +import java.util.Optional; +import tech.pegasys.teku.spec.datastructures.blobs.versions.fulu.DataColumnSidecar; +import tech.pegasys.teku.statetransition.validation.InternalValidationResult; + +public interface DasGossipLogger extends SubnetGossipLogger { + + DasGossipLogger NOOP = + new DasGossipLogger() { + @Override + public void onReceive( + DataColumnSidecar sidecar, InternalValidationResult validationResult) {} + + @Override + public void onPublish(DataColumnSidecar sidecar, Optional result) {} + + @Override + public void onDataColumnSubnetSubscribe(int subnetId) {} + + @Override + public void onDataColumnSubnetUnsubscribe(int subnetId) {} + }; +} diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/GossipLogger.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/GossipLogger.java new file mode 100644 index 00000000000..c1346155e2d --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/GossipLogger.java @@ -0,0 +1,24 @@ +/* + * 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.statetransition.datacolumns.log.gossip; + +import java.util.Optional; +import tech.pegasys.teku.statetransition.validation.InternalValidationResult; + +public interface GossipLogger { + + void onReceive(TMessage message, InternalValidationResult validationResult); + + void onPublish(TMessage message, Optional result); +} diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/SubnetGossipLogger.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/SubnetGossipLogger.java new file mode 100644 index 00000000000..f12a34ab6d4 --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/log/gossip/SubnetGossipLogger.java @@ -0,0 +1,21 @@ +/* + * 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.statetransition.datacolumns.log.gossip; + +public interface SubnetGossipLogger extends GossipLogger { + + void onDataColumnSubnetSubscribe(int subnetId); + + void onDataColumnSubnetUnsubscribe(int subnetId); +} diff --git a/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtil.java b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtil.java new file mode 100644 index 00000000000..c3243572967 --- /dev/null +++ b/ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtil.java @@ -0,0 +1,139 @@ +/* + * 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.datacolumns.util; + +import static java.lang.Integer.max; +import static java.lang.Integer.min; + +import java.util.BitSet; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import java.util.stream.Stream; +import org.apache.tuweni.bytes.Bytes; + +public class StringifyUtil { + + public static String columnIndexesToString( + final Collection indexes, final int maxColumns) { + final String lenStr = "(len: " + indexes.size() + ") "; + if (indexes.isEmpty()) { + return lenStr + "[]"; + } else if (indexes.size() == maxColumns) { + return lenStr + "[all]"; + } else if (maxColumns - indexes.size() <= 16) { + final Set exceptIndexes = + IntStream.range(0, maxColumns).boxed().collect(Collectors.toSet()); + exceptIndexes.removeAll(indexes); + return lenStr + "[all except " + sortAndJoin(exceptIndexes) + "]"; + } else { + final List ranges = reduceToIntRanges(indexes); + if (ranges.size() <= 16) { + return lenStr + + "[" + + ranges.stream().map(Objects::toString).collect(Collectors.joining(",")) + + "]"; + } else { + final BitSet bitSet = new BitSet(maxColumns); + indexes.forEach(bitSet::set); + return lenStr + "[bitmap: " + Bytes.of(bitSet.toByteArray()) + "]"; + } + } + } + + public static String toIntRangeStringWithSize(final Collection ints) { + return "(size: " + ints.size() + ") " + toIntRangeString(ints); + } + + public static String toIntRangeString(final Collection ints) { + final List ranges = reduceToIntRanges(ints); + return "[" + ranges.stream().map(Objects::toString).collect(Collectors.joining(",")) + "]"; + } + + private record IntRange(int first, int last) { + + static IntRange of(final int i) { + return new IntRange(i, i); + } + + static List union(final List left, final List right) { + if (left.isEmpty()) { + return right; + } else if (right.isEmpty()) { + return right; + } else { + return Stream.of( + left.stream().limit(left.size() - 1), + left.getLast().union(right.getFirst()).stream(), + right.stream().skip(1)) + .flatMap(s -> s) + .toList(); + } + } + + boolean isEmpty() { + return first > last; + } + + @SuppressWarnings("UnusedMethod") + boolean isSingle() { + return first == last; + } + + int size() { + return Integer.max(0, last() - first() + 1); + } + + List union(final IntRange other) { + if (this.isEmpty()) { + return List.of(other); + } else if (other.isEmpty()) { + return List.of(this); + } else if (other.first() > this.last() + 1) { + return List.of(this, other); + } else if (this.first() > other.last() + 1) { + return List.of(other, this); + } else { + return List.of( + new IntRange(min(this.first(), other.first()), max(this.last(), other.last()))); + } + } + + @Override + public String toString() { + return switch (size()) { + case 0 -> ""; + case 1 -> Integer.toString(first()); + case 2 -> first() + "," + last(); + default -> first() + ".." + last(); + }; + } + } + + private static List reduceToIntRanges(final Collection nums) { + return nums.stream() + .sorted() + .map(i -> List.of(IntRange.of(i))) + .reduce(IntRange::union) + .orElse(Collections.emptyList()); + } + + private static String sortAndJoin(final Collection nums) { + return nums.stream().sorted().map(Objects::toString).collect(Collectors.joining(",")); + } +} diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtilTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtilTest.java new file mode 100644 index 00000000000..1fd2f5936de --- /dev/null +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/datacolumns/util/StringifyUtilTest.java @@ -0,0 +1,71 @@ +/* + * 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.datacolumns.util; + +import static java.util.stream.IntStream.range; +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import java.util.Set; +import java.util.stream.IntStream; +import java.util.stream.Stream; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +public class StringifyUtilTest { + + record TestCase(IntStream indexes, String expectedString) { + @Override + public String toString() { + return expectedString; + } + } + + static final int MAX_INDEXES_LEN = 128; + + static final List TEST_CASES = + List.of( + new TestCase(IntStream.empty(), "[]"), + new TestCase(range(0, 128), "[all]"), + new TestCase(range(0, 128).skip(1), "[all except 0]"), + new TestCase(range(0, 128).limit(127), "[all except 127]"), + new TestCase( + range(0, 128).filter(i -> !Set.of(14, 16, 18, 98).contains(i)), + "[all except 14,16,18,98]"), + new TestCase(IntStream.of(14, 16, 18, 98), "[14,16,18,98]"), + new TestCase(IntStream.of(14, 15, 18, 98), "[14,15,18,98]"), + new TestCase(IntStream.of(14, 15, 16, 98), "[14..16,98]"), + new TestCase(range(0, 128).skip(64), "[64..127]"), + new TestCase(range(0, 128).limit(64), "[0..63]"), + new TestCase( + range(0, 128).filter(i -> i % 3 != 0), + "[bitmap: 0xb66ddbb66ddbb66ddbb66ddbb66ddbb6]"), + new TestCase( + range(10, 100).filter(i -> !Set.of(14, 16, 18, 98).contains(i)), + "[10..13,15,17,19..97,99]")); + + private static Stream provideTestCaseParameters() { + return TEST_CASES.stream().map(Arguments::of); + } + + @ParameterizedTest + @MethodSource("provideTestCaseParameters") + void columnIndexesToString_test(final TestCase testCase) { + final List idxList = testCase.indexes.boxed().toList(); + final String s = StringifyUtil.columnIndexesToString(idxList, MAX_INDEXES_LEN); + + assertThat(s).isEqualTo("(len: " + idxList.size() + ") " + testCase.expectedString); + } +} diff --git a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/util/BlockBlobSidecarsTrackersPoolImplTest.java b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/util/BlockBlobSidecarsTrackersPoolImplTest.java index a5c86cd8b49..ad0610b9f5c 100644 --- a/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/util/BlockBlobSidecarsTrackersPoolImplTest.java +++ b/ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/util/BlockBlobSidecarsTrackersPoolImplTest.java @@ -656,7 +656,7 @@ void shouldFetchMissingBlobSidecarsFromLocalELFirst() { .getKZGCommitment() .getBytesCompressed()) .kzgCommitmentInclusionProof( - miscHelpersDeneb.computeKzgCommitmentInclusionProof( + miscHelpersDeneb.computeBlobKzgCommitmentInclusionProof( index, block.getMessage().getBody())) .build()) .toList();