[WIP] Reconstruct supernode sidecars extension

eth-benchmark-tests/src/jmh/java/tech/pegasys/teku/benchmarks/kzg/WithPrecomputeBenchmark.java

@@ -13,7 +13,10 @@
 
 package tech.pegasys.teku.benchmarks.kzg;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import org.apache.tuweni.bytes.Bytes;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.Fork;
 import org.openjdk.jmh.annotations.Level;
@@ -23,6 +26,12 @@
 import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
 import org.openjdk.jmh.annotations.Warmup;
+import tech.pegasys.teku.kzg.KZGProof;
+import tech.pegasys.teku.spec.datastructures.blobs.DataColumnSidecar;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSchema;
+import tech.pegasys.teku.spec.datastructures.execution.BlobAndCellProofs;
+import tech.pegasys.teku.spec.datastructures.type.SszKZGProof;
+import tech.pegasys.teku.statetransition.datacolumns.DataColumnSidecarArchiveReconstructor;
 
 @Fork(1)
 @Warmup(iterations = 5, time = 2000, timeUnit = TimeUnit.MILLISECONDS)
@@ -64,4 +73,39 @@ public void reconstructDataColumnSidecars(final ExecutionPlan plan) {
     plan.config.miscHelpersFulu.reconstructAllDataColumnSidecars(
         plan.config.dataColumnSidecars.subList(halfSize, size));
   }
+
+  /** Similar to {@link DataColumnSidecarArchiveReconstructor} */
+  @Benchmark
+  public void reconstructDataColumnSidecarsFromArchive(final ExecutionPlan plan) {
+    final BlobSchema blobSchema = plan.config.schemaDefinitionsFulu.getBlobSchema();
+    final int size = plan.config.dataColumnSidecars.size();
+    final int halfSize = size / 2;
+    final List<DataColumnSidecar> sidecars = plan.config.dataColumnSidecars.subList(0, halfSize);
+
+    final List<List<KZGProof>> kzgProofs =
+        plan.config.dataColumnSidecars.subList(halfSize, size).stream()
+            .map(sidecar -> sidecar.getKzgProofs().stream().map(SszKZGProof::getKZGProof).toList())
+            .toList();
+    final List<BlobAndCellProofs> blobAndCellProofsList = new ArrayList<>();
+    for (int i = 0; i < sidecars.getFirst().getKzgCommitments().size(); i++) {
+      final int blobIndex = i;
+      final Bytes blob =
+          sidecars.stream()
+              .map(sidecar -> sidecar.getColumn().get(blobIndex).getBytes())
+              .reduce(Bytes.EMPTY, Bytes::concatenate);
+      final List<KZGProof> blobProofs = new ArrayList<>();
+      for (int j = 0; j < halfSize; j++) {
+        blobProofs.add(sidecars.get(j).getKzgProofs().get(blobIndex).getKZGProof());
+      }
+      for (int j = 0; j < halfSize; j++) {
+        blobProofs.add(kzgProofs.get(j).get(blobIndex));
+      }
+      final BlobAndCellProofs blobAndCellProofs =
+          new BlobAndCellProofs(blobSchema.create(blob), blobProofs);
+      blobAndCellProofsList.add(blobAndCellProofs);
+    }
+
+    plan.config.miscHelpersFulu.constructDataColumnSidecars(
+        plan.config.signedBeaconBlock, blobAndCellProofsList);
+  }
 }

ethereum/networks/src/main/java/tech/pegasys/teku/networks/Eth2NetworkConfiguration.java

@@ -103,6 +103,8 @@ public class Eth2NetworkConfiguration {
   public static final int DEFAULT_ASYNC_BEACON_CHAIN_MAX_THREADS =
       Math.max(Runtime.getRuntime().availableProcessors(), DEFAULT_VALIDATOR_EXECUTOR_THREADS);
 
+  public static final int DEFAULT_DATA_COLUMN_SIDECAR_EXTENSION_RETENTION_EPOCHS = 512;
+
   public static final int DEFAULT_ASYNC_BEACON_CHAIN_MAX_QUEUE = DEFAULT_MAX_QUEUE_SIZE;
 
   public static final String FINALIZED_STATE_URL_PATH = "eth/v2/debug/beacon/states/finalized";
@@ -150,6 +152,7 @@ public class Eth2NetworkConfiguration {
   private final int aggregatingAttestationPoolV2BlockAggregationTimeLimit;
   private final int aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit;
   private final int attestationWaitLimitMillis;
+  private final int dataColumnSidecarExtensionRetentionEpochs;
 
   private Eth2NetworkConfiguration(
       final Spec spec,
@@ -188,7 +191,8 @@ private Eth2NetworkConfiguration(
       final boolean aggregatingAttestationPoolProfilingEnabled,
       final int aggregatingAttestationPoolV2BlockAggregationTimeLimit,
       final int aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit,
-      final int attestationWaitLimitMillis) {
+      final int attestationWaitLimitMillis,
+      final int dataColumnSidecarExtensionRetentionEpochs) {
     this.spec = spec;
     this.constants = constants;
     this.stateBoostrapConfig = stateBoostrapConfig;
@@ -232,6 +236,7 @@ private Eth2NetworkConfiguration(
     this.aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit =
         aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit;
     this.attestationWaitLimitMillis = attestationWaitLimitMillis;
+    this.dataColumnSidecarExtensionRetentionEpochs = dataColumnSidecarExtensionRetentionEpochs;
 
     LOG.debug("Attestation wait time limit in ratchet: {} ms", attestationWaitLimitMillis);
 
@@ -393,6 +398,10 @@ public OptionalLong getDataColumnSidecarRecoveryMaxDelayMillis() {
     return dataColumnSidecarRecoveryMaxDelayMillis;
   }
 
+  public int getDataColumnSidecarExtensionRetentionEpochs() {
+    return dataColumnSidecarExtensionRetentionEpochs;
+  }
+
   @Override
   public String toString() {
     return constants;
@@ -424,6 +433,8 @@ public boolean equals(final Object o) {
             == that.aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit
         && forkChoiceUpdatedAlwaysSendPayloadAttributes
             == that.forkChoiceUpdatedAlwaysSendPayloadAttributes
+        && dataColumnSidecarExtensionRetentionEpochs
+            == that.dataColumnSidecarExtensionRetentionEpochs
         && rustKzgEnabled == that.rustKzgEnabled
         && Objects.equals(spec, that.spec)
         && Objects.equals(constants, that.constants)
@@ -477,7 +488,8 @@ public int hashCode() {
         forkChoiceLateBlockReorgEnabled,
         prepareBlockProductionEnabled,
         forkChoiceUpdatedAlwaysSendPayloadAttributes,
-        rustKzgEnabled);
+        rustKzgEnabled,
+        dataColumnSidecarExtensionRetentionEpochs);
   }
 
   public static class Builder {
@@ -520,6 +532,8 @@ public static class Builder {
     private boolean rustKzgEnabled = DEFAULT_RUST_KZG_ENABLED;
     private OptionalInt kzgPrecompute = OptionalInt.empty();
     private OptionalLong dataColumnSidecarRecoveryMaxDelayMillis = OptionalLong.empty();
+    private int dataColumnSidecarExtensionRetentionEpochs =
+        DEFAULT_DATA_COLUMN_SIDECAR_EXTENSION_RETENTION_EPOCHS;
     private boolean strictConfigLoadingEnabled;
     private boolean aggregatingAttestationPoolV2Enabled =
         DEFAULT_AGGREGATING_ATTESTATION_POOL_V2_ENABLED;
@@ -629,7 +643,8 @@ public Eth2NetworkConfiguration build() {
           aggregatingAttestationPoolProfilingEnabled,
           aggregatingAttestationPoolV2BlockAggregationTimeLimit,
           aggregatingAttestationPoolV2TotalBlockAggregationTimeLimit,
-          attestationWaitLimitMillis);
+          attestationWaitLimitMillis,
+          dataColumnSidecarExtensionRetentionEpochs);
     }
 
     private void validateCommandLineParameters() {
@@ -939,6 +954,15 @@ public Builder dataColumnSidecarRecoveryMaxDelayMillis(
       return this;
     }
 
+    public Builder dataColumnSidecarExtensionRetentionEpochs(
+        final int dataColumnSidecarExtensionRetentionEpochs) {
+      checkArgument(
+          dataColumnSidecarExtensionRetentionEpochs >= 0,
+          "Negative number of epochs is not allowed");
+      this.dataColumnSidecarExtensionRetentionEpochs = dataColumnSidecarExtensionRetentionEpochs;
+      return this;
+    }
+
     public Builder applyNetworkDefaults(final String networkName) {
       checkNotNull(networkName);
       Eth2Network.fromStringLenient(networkName)

ethereum/spec/src/main/java/tech/pegasys/teku/spec/Spec.java

@@ -27,6 +27,7 @@
 import it.unimi.dsi.fastutil.ints.IntList;
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumMap;
@@ -38,11 +39,14 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 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 org.apache.tuweni.bytes.Bytes48;
+import org.apache.tuweni.ssz.SSZ;
 import tech.pegasys.teku.bls.BLSPublicKey;
 import tech.pegasys.teku.bls.BLSSignatureVerifier;
 import tech.pegasys.teku.ethereum.performance.trackers.BlockProductionPerformance;
@@ -52,6 +56,7 @@
 import tech.pegasys.teku.infrastructure.ssz.SszList;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.kzg.KZG;
+import tech.pegasys.teku.kzg.KZGProof;
 import tech.pegasys.teku.spec.cache.IndexedAttestationCache;
 import tech.pegasys.teku.spec.config.NetworkingSpecConfig;
 import tech.pegasys.teku.spec.config.NetworkingSpecConfigDeneb;
@@ -98,6 +103,7 @@
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.common.BeaconStateInvariants;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.versions.altair.BeaconStateAltair;
+import tech.pegasys.teku.spec.datastructures.type.SszKZGProof;
 import tech.pegasys.teku.spec.datastructures.util.AttestationProcessingResult;
 import tech.pegasys.teku.spec.datastructures.util.ForkAndSpecMilestone;
 import tech.pegasys.teku.spec.genesis.GenesisGenerator;
@@ -540,6 +546,33 @@ public DataColumnSidecar deserializeSidecar(final Bytes serializedSidecar, final
         .sszDeserialize(serializedSidecar);
   }
 
+  public List<List<KZGProof>> deserializeProofs(final Bytes serializedProofs, final UInt64 slot) {
+    return SSZ.decode(
+        serializedProofs,
+        reader -> {
+          final List<List<KZGProof>> output = new ArrayList<>();
+          while (!reader.isComplete()) {
+            final List<Bytes> bytes = reader.readFixedBytesList(KZGProof.SIZE);
+            output.add(
+                bytes.stream()
+                    .map(proofBytes -> new KZGProof(Bytes48.wrap(proofBytes)))
+                    .collect(Collectors.toList()));
+          }
+          return output;
+        });
+  }
+
+  public Bytes serializeProofs(final List<DataColumnSidecar> dataColumnSidecars) {
+    return SSZ.encode(
+        writer ->
+            dataColumnSidecars.forEach(
+                dataColumnSidecar ->
+                    writer.writeFixedBytesList(
+                        dataColumnSidecar.getKzgProofs().stream()
+                            .map(SszKZGProof::getBytes)
+                            .toList())));
+  }
+
   // BeaconState
   public UInt64 getCurrentEpoch(final BeaconState state) {
     return atState(state).beaconStateAccessors().getCurrentEpoch(state);

ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/fulu/helpers/MiscHelpersFulu.java

@@ -183,6 +183,10 @@ public List<UInt64> computeColumnsForCustodyGroup(final UInt64 custodyGroup) {
         .toList();
   }
 
+  public List<List<KZGCell>> computeCells(final List<Blob> blobs) {
+    return blobs.parallelStream().map(blob -> getKzg().computeCells(blob.getBytes())).toList();
+  }
+
   private UInt64 computeCustodyGroupIndex(final UInt256 nodeId) {
     return bytesToUInt64(Hash.sha256(uint256ToBytes(nodeId)).slice(0, 8))
         .mod(specConfigFulu.getNumberOfCustodyGroups());

ethereum/spec/src/test/java/tech/pegasys/teku/spec/SpecTest.java

@@ -17,8 +17,15 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.IntStream;
+import org.apache.tuweni.bytes.Bytes;
 import org.junit.jupiter.api.Test;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.kzg.KZGProof;
+import tech.pegasys.teku.spec.datastructures.blobs.DataColumnSidecar;
+import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlockHeader;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBlockAndState;
+import tech.pegasys.teku.spec.datastructures.type.SszKZGProof;
 import tech.pegasys.teku.spec.generator.ChainBuilder;
 import tech.pegasys.teku.spec.util.DataStructureUtil;
 import tech.pegasys.teku.storage.server.StateStorageMode;
@@ -45,4 +52,28 @@ void shouldWindStateForwardIfOutsidePeriod() {
                 chain.getLast().getState(), dataStructureUtil.randomAttestation(100)))
         .isEqualTo(48);
   }
+
+  @Test
+  void shouldRoundTripKzgProofs() {
+    final Spec spec = TestSpecFactory.createMinimalFulu();
+    final DataStructureUtil dataStructureUtil = new DataStructureUtil(spec);
+    final SignedBeaconBlockHeader header = dataStructureUtil.randomSignedBeaconBlockHeader();
+    final List<DataColumnSidecar> dataColumnSidecars =
+        IntStream.range(64, 128)
+            .mapToObj(
+                index -> dataStructureUtil.randomDataColumnSidecar(header, UInt64.valueOf(index)))
+            .toList();
+    final List<List<KZGProof>> expectedProofs =
+        dataColumnSidecars.stream()
+            .map(
+                dataColumnSidecar ->
+                    dataColumnSidecar.getKzgProofs().stream()
+                        .map(SszKZGProof::getKZGProof)
+                        .toList())
+            .toList();
+    final Bytes serializedProofs = spec.serializeProofs(dataColumnSidecars);
+    final List<List<KZGProof>> kzgProofs =
+        spec.deserializeProofs(serializedProofs, header.getMessage().getSlot());
+    assertThat(expectedProofs).isEqualTo(kzgProofs);
+  }
 }

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/datacolumns/DataColumnSidecarArchiveReconstructor.java

@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import java.util.Optional;
+import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.datastructures.blobs.DataColumnSidecar;
+import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
+import tech.pegasys.teku.spec.datastructures.state.Checkpoint;
+import tech.pegasys.teku.storage.api.FinalizedCheckpointChannel;
+
+public interface DataColumnSidecarArchiveReconstructor extends FinalizedCheckpointChannel {
+
+  DataColumnSidecarArchiveReconstructor NOOP =
+      new DataColumnSidecarArchiveReconstructor() {
+        @Override
+        public SafeFuture<Optional<DataColumnSidecar>> reconstructDataColumnSidecar(
+            final SignedBeaconBlock block, final UInt64 index, final Bytes32 requestHash) {
+          return SafeFuture.completedFuture(Optional.empty());
+        }
+
+        @Override
+        public boolean isSidecarPruned(final UInt64 slot, final UInt64 index) {
+          return false;
+        }
+
+        @Override
+        public void onRequestCompleted(final Bytes32 requestHash) {}
+
+        @Override
+        public void onNewFinalizedCheckpoint(
+            final Checkpoint checkpoint, final boolean fromOptimisticBlock) {}
+      };
+
+  SafeFuture<Optional<DataColumnSidecar>> reconstructDataColumnSidecar(
+      SignedBeaconBlock block, UInt64 index, Bytes32 requestHash);
+
+  boolean isSidecarPruned(UInt64 slot, UInt64 index);
+
+  void onRequestCompleted(Bytes32 requestHash);
+}