Merge branch 'master' into draft-fc-generated-integration-3

Author: lucassaldanha

.github/workflows/ci.yml

@@ -260,7 +260,7 @@ jobs:
 
   integrationTests:
     needs: assemble
-    runs-on: ubuntu-latest-128
+    runs-on: gha-runner-scale-set-ubuntu-22.04-amd64-xxl
     steps:
       - name: Checkout
         uses: actions/checkout@v5
@@ -317,7 +317,7 @@ jobs:
       gradle_task: acceptanceTest
       src_pattern: "*/src/acceptance-test/java/*"
       src_root: "src/acceptance-test/java"
-      runner: "ubuntu-latest-128"
+      runner: "gha-runner-scale-set-ubuntu-22.04-amd64-xxl"
 
   acceptanceTestsReport:
     runs-on: ubuntu-24.04

eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/phase0/forkchoice/ForkChoiceTestExecutor.java

@@ -76,8 +76,6 @@
 import tech.pegasys.teku.statetransition.datacolumns.DasCustodyStand;
 import tech.pegasys.teku.statetransition.datacolumns.DasSamplerBasic;
 import tech.pegasys.teku.statetransition.datacolumns.DataColumnSidecarRecoveringCustody;
-import tech.pegasys.teku.statetransition.datacolumns.db.DataColumnSidecarDB;
-import tech.pegasys.teku.statetransition.datacolumns.db.DataColumnSidecarDbAccessor;
 import tech.pegasys.teku.statetransition.datacolumns.retriever.DataColumnSidecarRetrieverStub;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoice;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoiceStateProvider;
@@ -153,21 +151,16 @@ spec, new SignedBlockAndState(anchorBlock, anchorState)),
     final StubBlobSidecarManager blobSidecarManager = new StubBlobSidecarManager(kzg);
     final CurrentSlotProvider currentSlotProvider =
         CurrentSlotProvider.create(spec, recentChainData.getStore());
-    final DataColumnSidecarDB sidecarDB =
-        DataColumnSidecarDB.create(
-            storageSystem.combinedChainDataClient(), storageSystem.chainStorage());
-    final DataColumnSidecarDbAccessor dbAccessor =
-        DataColumnSidecarDbAccessor.builder(sidecarDB).spec(spec).build();
     final DasSamplerBasic dasSampler =
         new DasSamplerBasic(
             spec,
             currentSlotProvider,
-            dbAccessor,
             DataColumnSidecarRecoveringCustody.NOOP,
             new DataColumnSidecarRetrieverStub(),
             // using a const for the custody group count here, the test doesn't care
             // and fetching from the config would break when not in fulu
-            () -> DasCustodyStand.createCustodyGroupCountManager(4, 8));
+            () -> DasCustodyStand.createCustodyGroupCountManager(4, 8),
+            recentChainData);
     final StubDataColumnSidecarManager dataColumnSidecarManager =
         new StubDataColumnSidecarManager(spec, recentChainData, dasSampler);
     // forkChoiceLateBlockReorgEnabled is true here always because this is the reference test

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/RemoteOrigin.java

@@ -18,5 +18,6 @@ public enum RemoteOrigin {
   GOSSIP,
   LOCAL_EL,
   LOCAL_PROPOSAL,
-  RECOVERED
+  RECOVERED,
+  CUSTODY
 }

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

@@ -16,20 +16,36 @@
 import static tech.pegasys.teku.statetransition.datacolumns.DataAvailabilitySampler.SamplingEligibilityStatus.REQUIRED;
 
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Supplier;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+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.blocks.SignedBeaconBlock;
+import tech.pegasys.teku.spec.datastructures.util.DataColumnSlotAndIdentifier;
+import tech.pegasys.teku.statetransition.blobs.RemoteOrigin;
 import tech.pegasys.teku.statetransition.datacolumns.util.StringifyUtil;
 
 public class DasPreSampler {
 
   private static final Logger LOG = LogManager.getLogger();
 
   private final DataAvailabilitySampler sampler;
+  private final DataColumnSidecarCustody custody;
+  private final Supplier<CustodyGroupCountManager> custodyGroupCountManagerSupplier;
 
-  public DasPreSampler(final DataAvailabilitySampler sampler) {
+  public DasPreSampler(
+      final DataAvailabilitySampler sampler,
+      final DataColumnSidecarCustody custody,
+      final Supplier<CustodyGroupCountManager> custodyGroupCountManagerSupplier) {
     this.sampler = sampler;
+    this.custody = custody;
+    this.custodyGroupCountManagerSupplier = custodyGroupCountManagerSupplier;
   }
 
   private boolean isSamplingRequired(final SignedBeaconBlock block) {
@@ -52,19 +68,52 @@ public void onNewPreImportBlocks(final Collection<SignedBeaconBlock> blocks) {
   }
 
   private void onNewPreImportBlock(final SignedBeaconBlock block) {
-    sampler
-        .checkDataAvailability(block.getSlot(), block.getRoot())
-        .finish(
-            succ ->
-                LOG.debug(
-                    "DasPreSampler: success pre-sampling block {} ({})",
-                    block.getSlot(),
-                    block.getRoot()),
-            err ->
-                LOG.debug(
-                    "DasPreSampler: error pre-sampling block {} ({}): {}",
-                    block.getSlot(),
-                    block.getRoot(),
-                    err));
+    final Set<DataColumnSlotAndIdentifier> requiredColumnIdentifiers =
+        new HashSet<>(calculateSamplingColumnIds(block.getSlot(), block.getRoot()));
+
+    final SafeFuture<List<DataColumnSlotAndIdentifier>> columnsInCustodyFuture =
+        maybeHasColumnsInCustody(requiredColumnIdentifiers);
+
+    columnsInCustodyFuture
+        .thenAccept(
+            columnsInCustody ->
+                columnsInCustody.forEach(
+                    columnId -> sampler.onAlreadyKnownDataColumn(columnId, RemoteOrigin.CUSTODY)))
+        .always(
+            () ->
+                sampler
+                    .checkDataAvailability(block.getSlot(), block.getRoot())
+                    .finish(
+                        succ ->
+                            LOG.debug(
+                                "DasPreSampler: success pre-sampling block {} ({})",
+                                block.getSlot(),
+                                block.getRoot()),
+                        err ->
+                            LOG.debug(
+                                "DasPreSampler: error pre-sampling block {} ({}): {}",
+                                block.getSlot(),
+                                block.getRoot(),
+                                err)));
+  }
+
+  private List<DataColumnSlotAndIdentifier> calculateSamplingColumnIds(
+      final UInt64 slot, final Bytes32 blockRoot) {
+    return custodyGroupCountManagerSupplier.get().getSamplingColumnIndices().stream()
+        .map(columnIndex -> new DataColumnSlotAndIdentifier(slot, blockRoot, columnIndex))
+        .toList();
+  }
+
+  private SafeFuture<Optional<DataColumnSlotAndIdentifier>> checkColumnInCustody(
+      final DataColumnSlotAndIdentifier columnIdentifier) {
+    return custody
+        .hasCustodyDataColumnSidecar(columnIdentifier)
+        .thenApply(hasColumn -> hasColumn ? Optional.of(columnIdentifier) : Optional.empty());
+  }
+
+  private SafeFuture<List<DataColumnSlotAndIdentifier>> maybeHasColumnsInCustody(
+      final Collection<DataColumnSlotAndIdentifier> columnIdentifiers) {
+    return SafeFuture.collectAll(columnIdentifiers.stream().map(this::checkColumnInCustody))
+        .thenApply(list -> list.stream().flatMap(Optional::stream).toList());
   }
 }