Created validatorTimingChannel entries for sync and contribution messages (#9822)

rolfyone · web-flow · commit 0b5abf5fc624 · 2025-08-29T10:42:30.000+10:00
partially addresses #9761
diff --git a/validator/api/src/main/java/tech/pegasys/teku/validator/api/ValidatorTimingChannel.java b/validator/api/src/main/java/tech/pegasys/teku/validator/api/ValidatorTimingChannel.java
@@ -41,6 +41,10 @@ void onHeadUpdate(
 
   void onAttestationAggregationDue(UInt64 slot);
 
+  void onSyncCommitteeCreationDue(UInt64 slot);
+
+  void onContributionCreationDue(UInt64 slot);
+
   void onAttesterSlashing(AttesterSlashing attesterSlashing);
 
   void onProposerSlashing(ProposerSlashing proposerSlashing);
diff --git a/validator/beaconnode/src/main/java/tech/pegasys/teku/validator/beaconnode/TimeBasedEventAdapter.java b/validator/beaconnode/src/main/java/tech/pegasys/teku/validator/beaconnode/TimeBasedEventAdapter.java
@@ -22,6 +22,7 @@
 import tech.pegasys.teku.infrastructure.time.TimeProvider;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.Spec;
+import tech.pegasys.teku.spec.SpecMilestone;
 import tech.pegasys.teku.validator.api.ValidatorTimingChannel;
 
 public class TimeBasedEventAdapter implements BeaconChainEventAdapter {
@@ -72,6 +73,16 @@ void start(final UInt64 genesisTime) {
         nextSlotStartTimeMillis.plus(spec.getAggregateDueMillis(nextSlot)),
         millisPerSlot,
         this::onAggregationDue);
+    if (spec.atSlot(nextSlot).getMilestone().isGreaterThanOrEqualTo(SpecMilestone.ALTAIR)) {
+      taskScheduler.scheduleRepeatingEventInMillis(
+          nextSlotStartTimeMillis.plus(spec.getSyncMessageDueMillis(nextSlot)),
+          millisPerSlot,
+          this::onSyncCommitteeCreationDue);
+      taskScheduler.scheduleRepeatingEventInMillis(
+          nextSlotStartTimeMillis.plus(spec.getContributionDueMillis(nextSlot)),
+          millisPerSlot,
+          this::onContributionCreationDue);
+    }
   }
 
   private UInt64 getCurrentSlot() {
@@ -99,6 +110,30 @@ private void onAttestationCreationDue(
     validatorTimingChannel.onAttestationCreationDue(slot);
   }
 
+  private void onSyncCommitteeCreationDue(
+      final UInt64 scheduledTimeInMillis, final UInt64 actualTimeInMillis) {
+    final UInt64 slot = spec.getCurrentSlotFromTimeMillis(scheduledTimeInMillis, genesisTimeMillis);
+    if (isTooLate(scheduledTimeInMillis, actualTimeInMillis)) {
+      LOG.warn(
+          "Skipping sync committee message for slot {} due to unexpected delay in slot processing",
+          slot);
+      return;
+    }
+    validatorTimingChannel.onSyncCommitteeCreationDue(slot);
+  }
+
+  private void onContributionCreationDue(
+      final UInt64 scheduledTimeInMillis, final UInt64 actualTimeInMillis) {
+    final UInt64 slot = spec.getCurrentSlotFromTimeMillis(scheduledTimeInMillis, genesisTimeMillis);
+    if (isTooLate(scheduledTimeInMillis, actualTimeInMillis)) {
+      LOG.warn(
+          "Skipping contribution message for slot {} due to unexpected delay in slot processing",
+          slot);
+      return;
+    }
+    validatorTimingChannel.onContributionCreationDue(slot);
+  }
+
   private void onAggregationDue(
       final UInt64 scheduledTimeInMillis, final UInt64 actualTimeInMillis) {
     final UInt64 slot = spec.getCurrentSlotFromTimeMillis(scheduledTimeInMillis, genesisTimeMillis);
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/AttestationDutyBatchSchedulingStrategy.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/AttestationDutyBatchSchedulingStrategy.java
@@ -183,6 +183,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/AttestationDutyScheduler.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/AttestationDutyScheduler.java
@@ -68,6 +68,12 @@ public void onAttestationCreationDue(final UInt64 slot) {
     onProductionDue(slot);
   }
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/BeaconProposerPreparer.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/BeaconProposerPreparer.java
@@ -95,6 +95,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/BlockDutyScheduler.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/BlockDutyScheduler.java
@@ -45,6 +45,12 @@ public void onBlockProductionDue(final UInt64 slot) {
     onProductionDue(slot);
   }
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/OwnedValidatorStatusProvider.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/OwnedValidatorStatusProvider.java
@@ -133,6 +133,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/SyncCommitteeScheduler.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/SyncCommitteeScheduler.java
@@ -127,7 +127,7 @@ private SyncCommitteePeriod createSyncCommitteePeriod(
   }
 
   @Override
-  public void onAttestationCreationDue(final UInt64 slot) {
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {
     // Check slot being null for the edge case of genesis slot (i.e. slot 0)
     if (lastProductionSlot != null && slot.compareTo(lastProductionSlot) <= 0) {
       LOG.debug(
@@ -142,7 +142,7 @@ public void onAttestationCreationDue(final UInt64 slot) {
   }
 
   @Override
-  public void onAttestationAggregationDue(final UInt64 slot) {
+  public void onContributionCreationDue(final UInt64 slot) {
     getDutiesForSlot(slot).ifPresent(duties -> duties.onAggregationDue(slot));
   }
 
@@ -189,6 +189,12 @@ public void onValidatorsAdded() {
   @Override
   public void onBlockProductionDue(final UInt64 slot) {}
 
+  @Override
+  public void onAttestationCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onAttestationAggregationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/ValidatorRegistrator.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/ValidatorRegistrator.java
@@ -128,6 +128,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/ValidatorTimingActions.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/ValidatorTimingActions.java
@@ -108,6 +108,16 @@ public void onAttestationAggregationDue(final UInt64 slot) {
     delegates.forEach(delegates -> delegates.onAttestationAggregationDue(slot));
   }
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {
+    delegates.forEach(delegate -> delegate.onSyncCommitteeCreationDue(slot));
+  }
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {
+    delegates.forEach(delegate -> delegate.onContributionCreationDue(slot));
+  }
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {
     delegates.forEach(delegates -> delegates.onAttesterSlashing(attesterSlashing));
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/duties/synccommittee/ChainHeadTracker.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/duties/synccommittee/ChainHeadTracker.java
@@ -69,6 +69,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
diff --git a/validator/client/src/main/java/tech/pegasys/teku/validator/client/loader/SlashingProtectionLogger.java b/validator/client/src/main/java/tech/pegasys/teku/validator/client/loader/SlashingProtectionLogger.java
@@ -188,6 +188,12 @@ public void onAttestationCreationDue(final UInt64 slot) {}
   @Override
   public void onAttestationAggregationDue(final UInt64 slot) {}
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onValidatorsAdded() {}
 
diff --git a/validator/client/src/test/java/tech/pegasys/teku/validator/client/SyncCommitteeSchedulerTest.java b/validator/client/src/test/java/tech/pegasys/teku/validator/client/SyncCommitteeSchedulerTest.java
@@ -96,7 +96,7 @@ void shouldPerformProductionForEachSlotWhenAttestationCreationDue() {
     UInt64.range(UInt64.ONE, UInt64.valueOf(10))
         .forEach(
             slot -> {
-              scheduler.onAttestationCreationDue(slot);
+              scheduler.onSyncCommitteeCreationDue(slot);
               verify(duties).performProductionDuty(slot);
             });
   }
@@ -107,8 +107,8 @@ void shouldNotPerformProductionMultipleTimesForSameSlot() {
 
     getRequestedDutiesForSyncCommitteePeriod(0).complete(Optional.of(duties));
 
-    scheduler.onAttestationCreationDue(UInt64.ONE);
-    scheduler.onAttestationCreationDue(UInt64.ONE);
+    scheduler.onSyncCommitteeCreationDue(UInt64.ONE);
+    scheduler.onSyncCommitteeCreationDue(UInt64.ONE);
 
     verify(duties, times(1)).performProductionDuty(UInt64.ONE);
   }
@@ -119,23 +119,23 @@ void shouldNotPerformProductionForEarlierSlot() {
 
     getRequestedDutiesForSyncCommitteePeriod(0).complete(Optional.of(duties));
 
-    scheduler.onAttestationCreationDue(UInt64.valueOf(2));
-    scheduler.onAttestationCreationDue(UInt64.ONE);
+    scheduler.onSyncCommitteeCreationDue(UInt64.valueOf(2));
+    scheduler.onSyncCommitteeCreationDue(UInt64.ONE);
 
     verify(duties).performProductionDuty(UInt64.valueOf(2));
     verify(duties, never()).performProductionDuty(UInt64.ONE);
   }
 
   @Test
-  void shouldPerformAggregationForEachSlotWhenAttestationAggregationDue() {
+  void shouldPerformAggregationForEachSlotWhenContributionDue() {
     scheduler.onSlot(UInt64.ONE);
 
     getRequestedDutiesForSyncCommitteePeriod(0).complete(Optional.of(duties));
 
     UInt64.range(UInt64.ONE, UInt64.valueOf(10))
         .forEach(
             slot -> {
-              scheduler.onAttestationAggregationDue(slot);
+              scheduler.onContributionCreationDue(slot);
               verify(duties).performAggregationDuty(slot);
             });
   }
@@ -250,12 +250,12 @@ void shouldSwitchToNextCommitteePeriodWhenLastSlotOfSyncCommitteePeriodReached()
     getRequestedDutiesForSyncCommitteePeriod(1).complete(Optional.of(nextDuties));
 
     // Subscribed, but still performing duties for first sync committee period
-    scheduler.onAttestationCreationDue(subscribeSlot);
+    scheduler.onSyncCommitteeCreationDue(subscribeSlot);
     verify(duties).performProductionDuty(subscribeSlot);
 
     final UInt64 nextSyncCommitteeFirstDutySlot = nextSyncCommitteePeriodStartSlot.minus(1);
     scheduler.onSlot(nextSyncCommitteeFirstDutySlot);
-    scheduler.onAttestationCreationDue(nextSyncCommitteeFirstDutySlot);
+    scheduler.onSyncCommitteeCreationDue(nextSyncCommitteeFirstDutySlot);
     verify(nextDuties).performProductionDuty(nextSyncCommitteeFirstDutySlot);
     verify(duties, never()).performProductionDuty(nextSyncCommitteeFirstDutySlot);
   }
@@ -281,8 +281,8 @@ void shouldUseNextPeriodForDutiesWhenOnSlotNotYetCalled() {
     getRequestedDutiesForSyncCommitteePeriod(1).complete(Optional.of(nextDuties));
 
     // Unexpectedly jump ahead to the next committee period without getting a slot event first
-    scheduler.onAttestationCreationDue(nextSyncCommitteePeriodStartSlot);
-    scheduler.onAttestationAggregationDue(nextSyncCommitteePeriodStartSlot);
+    scheduler.onSyncCommitteeCreationDue(nextSyncCommitteePeriodStartSlot);
+    scheduler.onContributionCreationDue(nextSyncCommitteePeriodStartSlot);
 
     // Should use duties from the next period
     verify(nextDuties).performProductionDuty(nextSyncCommitteePeriodStartSlot);
@@ -293,8 +293,8 @@ void shouldUseNextPeriodForDutiesWhenOnSlotNotYetCalled() {
         syncCommitteeUtil.computeFirstEpochOfNextSyncCommitteePeriod(
             nextSyncCommitteePeriodStartEpoch);
     final UInt64 tooFarInFutureSlot = spec.computeStartSlotAtEpoch(tooFarInFutureEpoch);
-    scheduler.onAttestationCreationDue(tooFarInFutureSlot);
-    scheduler.onAttestationAggregationDue(tooFarInFutureSlot);
+    scheduler.onSyncCommitteeCreationDue(tooFarInFutureSlot);
+    scheduler.onContributionCreationDue(tooFarInFutureSlot);
 
     verify(duties, never()).performProductionDuty(tooFarInFutureSlot);
     verify(duties, never()).performAggregationDuty(tooFarInFutureSlot);
diff --git a/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/BeaconNodeReadinessManager.java b/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/BeaconNodeReadinessManager.java
@@ -134,6 +134,12 @@ public void onAttestationAggregationDue(final UInt64 slot) {
     performReadinessCheckAgainstAllNodes().finishError(LOG);
   }
 
+  @Override
+  public void onSyncCommitteeCreationDue(final UInt64 slot) {}
+
+  @Override
+  public void onContributionCreationDue(final UInt64 slot) {}
+
   @Override
   public void onAttesterSlashing(final AttesterSlashing attesterSlashing) {}
 
