Added more utilities for late block reorg

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/block/BlockManager.java

@@ -132,6 +132,9 @@ public SafeFuture<InternalValidationResult> validateAndImportBlock(
 
     final Optional<BlockImportPerformance> blockImportPerformance;
 
+    arrivalTimestamp.ifPresent(
+        arrivalTime -> recentChainData.setBlockTimelinessFromArrivalTime(block, arrivalTime));
+
     if (blockImportMetrics.isPresent()) {
       final BlockImportPerformance performance =
           new BlockImportPerformance(timeProvider, blockImportMetrics.get());

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceNotifier.java

@@ -32,6 +32,8 @@ SafeFuture<Optional<ExecutionPayloadContext>> getPayloadId(
 
   void onTerminalBlockReached(Bytes32 executionBlockHash);
 
+  boolean validatorIsConnected(UInt64 validatorIndex, UInt64 currentSlot);
+
   long subscribeToForkChoiceUpdatedResult(ForkChoiceUpdatedResultSubscriber subscriber);
 
   boolean unsubscribeFromForkChoiceUpdatedResult(long subscriberId);

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceNotifierImpl.java

@@ -108,6 +108,11 @@ public void onTerminalBlockReached(Bytes32 executionBlockHash) {
     eventThread.execute(() -> internalTerminalBlockReached(executionBlockHash));
   }
 
+  @Override
+  public boolean validatorIsConnected(UInt64 validatorIndex, UInt64 currentSlot) {
+    return proposersDataManager.validatorIsConnected(validatorIndex, currentSlot);
+  }
+
   @Override
   public void onPreparedProposersUpdated() {
     eventThread.execute(this::internalUpdatePreparableProposers);

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/ProposersDataManager.java

@@ -139,6 +139,12 @@ public SafeFuture<Void> updateValidatorRegistrations(
                     headState, signedValidatorRegistrations, currentSlot));
   }
 
+  // used in ForkChoice validator_is_connected
+  public boolean validatorIsConnected(final UInt64 validatorIndex, final UInt64 currentSlot) {
+    final PreparedProposerInfo info = preparedProposerInfoByValidatorIndex.get(validatorIndex);
+    return info != null && !info.hasExpired(currentSlot);
+  }
+
   private void updatePreparedProposerCache(
       final Collection<BeaconPreparableProposer> preparedProposers, final UInt64 currentSlot) {
     final UInt64 expirySlot =

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ProposersDataManagerTest.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright Consensys Software Inc., 2023
+ *
+ * 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.forkchoice;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import java.util.Optional;
+import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.junit.jupiter.api.Test;
+import tech.pegasys.teku.ethereum.execution.types.Eth1Address;
+import tech.pegasys.teku.infrastructure.async.eventthread.EventThread;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.spec.Spec;
+import tech.pegasys.teku.spec.TestSpecFactory;
+import tech.pegasys.teku.spec.datastructures.operations.versions.bellatrix.BeaconPreparableProposer;
+import tech.pegasys.teku.spec.executionlayer.ExecutionLayerChannel;
+import tech.pegasys.teku.spec.util.DataStructureUtil;
+import tech.pegasys.teku.storage.client.RecentChainData;
+
+class ProposersDataManagerTest {
+
+  private final Spec spec = TestSpecFactory.createMinimalCapella();
+
+  private final DataStructureUtil dataStructureUtil = new DataStructureUtil(spec);
+
+  private final RecentChainData recentChainData = mock(RecentChainData.class);
+
+  private final ExecutionLayerChannel channel = ExecutionLayerChannel.NOOP;
+  private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
+
+  private final Eth1Address defaultAddress = dataStructureUtil.randomEth1Address();
+  private final ProposersDataManager manager =
+      new ProposersDataManager(
+          mock(EventThread.class),
+          spec,
+          metricsSystem,
+          channel,
+          recentChainData,
+          Optional.of(defaultAddress));
+
+  final List<BeaconPreparableProposer> proposers =
+      List.of(
+          new BeaconPreparableProposer(UInt64.ONE, dataStructureUtil.randomEth1Address()),
+          new BeaconPreparableProposer(UInt64.ZERO, defaultAddress));
+
+  @Test
+  void validatorIsConnected_notFound_withEmptyPreparedList() {
+    assertThat(manager.validatorIsConnected(UInt64.ZERO, UInt64.ZERO)).isFalse();
+  }
+
+  @Test
+  void validatorIsConnected_found_withPreparedProposer() {
+    manager.updatePreparedProposers(proposers, UInt64.ONE);
+    assertThat(manager.validatorIsConnected(UInt64.ONE, UInt64.valueOf(1))).isTrue();
+  }
+
+  @Test
+  void validatorIsConnected_notFound_withDifferentPreparedProposer() {
+    manager.updatePreparedProposers(proposers, UInt64.ONE);
+    assertThat(manager.validatorIsConnected(UInt64.valueOf(2), UInt64.valueOf(2))).isFalse();
+  }
+
+  @Test
+  void validatorIsConnected_notFound_withExpiredPreparedProposer() {
+    manager.updatePreparedProposers(proposers, UInt64.ONE);
+    assertThat(manager.validatorIsConnected(UInt64.ONE, UInt64.valueOf(26))).isFalse();
+  }
+}

ethereum/statetransition/src/testFixtures/java/tech/pegasys/teku/statetransition/forkchoice/StubForkChoiceNotifier.java

@@ -69,4 +69,9 @@ public SafeFuture<Optional<ExecutionPayloadContext>> getPayloadId(
 
   @Override
   public void onTerminalBlockReached(final Bytes32 executionBlockHash) {}
+
+  @Override
+  public boolean validatorIsConnected(UInt64 validatorIndex, UInt64 currentSlot) {
+    return true;
+  }
 }

services/beaconchain/src/main/java/tech/pegasys/teku/services/beaconchain/BeaconChainController.java

@@ -401,6 +401,7 @@ protected SafeFuture<?> initialize() {
                     metricsSystem,
                     storeConfig,
                     beaconAsyncRunner,
+                    timeProvider,
                     storageQueryChannel,
                     storageUpdateChannel,
                     voteUpdateChannel,

storage/src/main/java/tech/pegasys/teku/storage/client/BlockTimelinessTracker.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright Consensys Software Inc., 2023
+ *
+ * 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.storage.client;
+
+import static tech.pegasys.teku.spec.constants.NetworkConstants.INTERVALS_PER_SLOT;
+
+import java.util.Map;
+import java.util.Optional;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.infrastructure.collections.LimitedMap;
+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.datastructures.blocks.SignedBeaconBlock;
+
+public class BlockTimelinessTracker {
+  private static final Logger LOG = LogManager.getLogger();
+  private final Map<Bytes32, Boolean> blockTimeliness;
+  private final TimeProvider timeProvider;
+  private final Spec spec;
+  private final RecentChainData recentChainData;
+
+  // implements is_timely from Consensus Spec
+  public BlockTimelinessTracker(
+      final Spec spec, final RecentChainData recentChainData, final TimeProvider timeProvider) {
+    this.spec = spec;
+    final int epochsForTimeliness =
+        Math.max(spec.getGenesisSpecConfig().getReorgMaxEpochsSinceFinalization(), 3);
+    this.blockTimeliness =
+        LimitedMap.createSynchronizedNatural(
+            spec.getGenesisSpec().getSlotsPerEpoch() * epochsForTimeliness);
+    this.timeProvider = timeProvider;
+    this.recentChainData = recentChainData;
+  }
+
+  public void setBlockTimelinessFromArrivalTime(
+      final SignedBeaconBlock block, final UInt64 arrivalTimeMillis) {
+    final UInt64 genesisTime = recentChainData.getGenesisTime();
+    final UInt64 computedSlot = spec.getCurrentSlot(timeProvider.getTimeInSeconds(), genesisTime);
+    final Bytes32 root = block.getRoot();
+    if (computedSlot.isGreaterThan(block.getMessage().getSlot())) {
+      LOG.debug(
+          "Block {}:{} is before computed slot {}, timeliness set to false.",
+          root,
+          block.getSlot(),
+          computedSlot);
+      blockTimeliness.put(root, false);
+      return;
+    }
+    recentChainData
+        .getCurrentSlot()
+        .ifPresent(
+            slot -> {
+              final UInt64 slotStartTimeMillis =
+                  spec.getSlotStartTimeMillis(slot, genesisTime.times(1000));
+              final int millisIntoSlot =
+                  arrivalTimeMillis.minusMinZero(slotStartTimeMillis).intValue();
+
+              final UInt64 timelinessLimit =
+                  spec.getMillisPerSlot(slot).dividedBy(INTERVALS_PER_SLOT);
+
+              final boolean isTimely =
+                  block.getMessage().getSlot().equals(slot)
+                      && timelinessLimit.isGreaterThan(millisIntoSlot);
+              LOG.debug(
+                  "Block {}:{} arrived at {} ms into slot {}, timeliness limit is {} ms. result: {}",
+                  root,
+                  block.getSlot(),
+                  millisIntoSlot,
+                  computedSlot,
+                  timelinessLimit,
+                  isTimely);
+              blockTimeliness.put(root, isTimely);
+            });
+  }
+
+  public Optional<Boolean> isBlockTimely(final Bytes32 root) {
+    return Optional.ofNullable(blockTimeliness.get(root));
+  }
+}

storage/src/main/java/tech/pegasys/teku/storage/client/RecentChainData.java

@@ -35,6 +35,7 @@
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.bytes.Bytes4;
 import tech.pegasys.teku.infrastructure.metrics.TekuMetricCategory;
+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;
@@ -98,10 +99,13 @@ public abstract class RecentChainData implements StoreUpdateHandler {
   private volatile Optional<ChainHead> chainHead = Optional.empty();
   private volatile UInt64 genesisTime;
 
+  private final BlockTimelinessTracker blockTimelinessTracker;
+
   RecentChainData(
       final AsyncRunner asyncRunner,
       final MetricsSystem metricsSystem,
       final StoreConfig storeConfig,
+      final TimeProvider timeProvider,
       final BlockProvider blockProvider,
       final StateAndBlockSummaryProvider stateProvider,
       final EarliestBlobSidecarSlotProvider earliestBlobSidecarSlotProvider,
@@ -120,6 +124,7 @@ public abstract class RecentChainData implements StoreUpdateHandler {
     this.chainHeadChannel = chainHeadChannel;
     this.storageUpdateChannel = storageUpdateChannel;
     this.finalizedCheckpointChannel = finalizedCheckpointChannel;
+    this.blockTimelinessTracker = new BlockTimelinessTracker(spec, this, timeProvider);
     reorgCounter =
         metricsSystem.createCounter(
             TekuMetricCategory.BEACON,
@@ -613,4 +618,13 @@ public List<Bytes32> getAllBlockRootsAtSlot(final UInt64 slot) {
         .map(forkChoiceStrategy -> forkChoiceStrategy.getBlockRootsAtSlot(slot))
         .orElse(Collections.emptyList());
   }
+
+  public void setBlockTimelinessFromArrivalTime(
+      final SignedBeaconBlock block, final UInt64 arrivalTime) {
+    blockTimelinessTracker.setBlockTimelinessFromArrivalTime(block, arrivalTime);
+  }
+
+  public Optional<Boolean> getBlockTimeliness(final Bytes32 blockRoot) {
+    return blockTimelinessTracker.isBlockTimely(blockRoot);
+  }
 }

storage/src/main/java/tech/pegasys/teku/storage/client/StorageBackedRecentChainData.java

@@ -26,6 +26,7 @@
 import tech.pegasys.teku.dataproviders.lookup.StateAndBlockSummaryProvider;
 import tech.pegasys.teku.infrastructure.async.AsyncRunner;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.time.TimeProvider;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.config.Constants;
 import tech.pegasys.teku.storage.api.ChainHeadChannel;
@@ -49,6 +50,7 @@ public StorageBackedRecentChainData(
       final AsyncRunner asyncRunner,
       final MetricsSystem metricsSystem,
       final StoreConfig storeConfig,
+      final TimeProvider timeProvider,
       final StorageQueryChannel storageQueryChannel,
       final StorageUpdateChannel storageUpdateChannel,
       final VoteUpdateChannel voteUpdateChannel,
@@ -59,6 +61,7 @@ public StorageBackedRecentChainData(
         asyncRunner,
         metricsSystem,
         storeConfig,
+        timeProvider,
         storageQueryChannel::getHotBlocksByRoot,
         storageQueryChannel::getHotStateAndBlockSummaryByBlockRoot,
         storageQueryChannel::getEarliestAvailableBlobSidecarSlot,
@@ -77,6 +80,7 @@ public static SafeFuture<RecentChainData> create(
       final MetricsSystem metricsSystem,
       final StoreConfig storeConfig,
       final AsyncRunner asyncRunner,
+      final TimeProvider timeProvider,
       final StorageQueryChannel storageQueryChannel,
       final StorageUpdateChannel storageUpdateChannel,
       final VoteUpdateChannel voteUpdateChannel,
@@ -88,6 +92,7 @@ public static SafeFuture<RecentChainData> create(
             asyncRunner,
             metricsSystem,
             storeConfig,
+            timeProvider,
             storageQueryChannel,
             storageUpdateChannel,
             voteUpdateChannel,
@@ -103,6 +108,7 @@ public static RecentChainData createImmediately(
       final AsyncRunner asyncRunner,
       final MetricsSystem metricsSystem,
       final StoreConfig storeConfig,
+      final TimeProvider timeProvider,
       final StorageQueryChannel storageQueryChannel,
       final StorageUpdateChannel storageUpdateChannel,
       final VoteUpdateChannel voteUpdateChannel,
@@ -114,6 +120,7 @@ public static RecentChainData createImmediately(
             asyncRunner,
             metricsSystem,
             storeConfig,
+            timeProvider,
             storageQueryChannel,
             storageUpdateChannel,
             voteUpdateChannel,