Das custody backfiller (DASCustodySync rewrite)

acceptance-tests/src/acceptance-test/java/tech/pegasys/teku/test/acceptance/das/DasCheckpointSyncAcceptanceTest.java

@@ -29,7 +29,7 @@
 import tech.pegasys.teku.test.acceptance.dsl.TekuBeaconNode;
 import tech.pegasys.teku.test.acceptance.dsl.TekuNodeConfigBuilder;
 
-@Disabled("until DataColumnSidecars backfill is fixed")
+@Disabled("until new DasCustodyBackfiller is enabled by default")
 public class DasCheckpointSyncAcceptanceTest extends AcceptanceTestBase {
   @Test
   public void shouldBeAbleToCheckpointSyncAndBackfillCustody() throws Exception {
@@ -72,13 +72,18 @@ public void shouldBeAbleToCheckpointSyncAndBackfillCustody() throws Exception {
 
     final UInt64 firstFuluSlot =
         primaryNode.getSpec().computeStartSlotAtEpoch(specConfigFulu.getFuluForkEpoch());
-    // Wait until block backfill is completed, it starts after forward sync
-    secondaryNode.waitForBlockAtSlot(firstFuluSlot);
+
+    // Wait until custody is backfilled is completed.
+    // It will require:
+    // - forward sync to catch up
+    // - backfill to complete
+    // - custody fully backfilled up to first fulu slot
+    final int expectedCustodyCount = specConfigFulu.getCustodyRequirement();
+    secondaryNode.waitForCustodyBackfill(firstFuluSlot, expectedCustodyCount);
 
     final SignedBeaconBlock blockAtHead = secondaryNode.getHeadBlock();
     final int checkpointSlot = checkpointFinalizedBlock.getSlot().intValue();
     final int endSlot = blockAtHead.getSlot().intValue();
-    final int expectedCustodyCount = specConfigFulu.getCustodyRequirement();
 
     // after checkpoint is synced with sidecars
     final int afterCheckpointSidecars =

acceptance-tests/src/testFixtures/java/tech/pegasys/teku/test/acceptance/dsl/TekuBeaconNode.java

@@ -739,6 +739,16 @@ public int getDataColumnSidecarCount(final String blockId) throws IOException {
     return jsonNode.get("data").size();
   }
 
+  public void waitForCustodyBackfill(final UInt64 slot, final int expectedCustodyCount) {
+    waitFor(
+        () -> {
+          final Optional<SignedBeaconBlock> maybeBlock = getBlockAtSlot(slot);
+          assertThat(maybeBlock).isPresent();
+          assertThat(getDataColumnSidecarCount(maybeBlock.get().getRoot().toHexString()))
+              .isEqualTo(expectedCustodyCount);
+        });
+  }
+
   public void waitForValidators(final int numberOfValidators) {
     waitFor(
         () -> {