Fix dropping from SYNCING to IN_SYNC while far behind the head

CHANGELOG.md

@@ -27,3 +27,4 @@
  - Trigger an immediate peer search when publishing sync committee messages fails because there are no peers available on the required gossip topic.
  - Fixed gossip wire validator to reject inbound messages containing the `key` field.
  - Fixed the gossip message size gate comparing the compressed payload size against the uncompressed `MAX_PAYLOAD_SIZE`.
+ - Fixed an out of memory error when a sync stopped while the chain head was still far behind. The node no longer reports itself as in sync in that case.

beacon/sync/build.gradle

@@ -1,4 +1,5 @@
 dependencies {
+	implementation project(':ethereum:events')
 	implementation project(':ethereum:spec')
 	implementation project(':ethereum:statetransition')
 	implementation project(':ethereum:executionclient')

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/DefaultSyncServiceFactory.java

@@ -31,6 +31,7 @@
 import tech.pegasys.teku.beacon.sync.gossip.blocks.RecentBlocksFetchService;
 import tech.pegasys.teku.beacon.sync.gossip.executionpayloads.RecentExecutionPayloadsFetcher;
 import tech.pegasys.teku.beacon.sync.historical.HistoricalBlockSyncService;
+import tech.pegasys.teku.ethereum.events.SlotEventsChannel;
 import tech.pegasys.teku.infrastructure.async.AsyncRunner;
 import tech.pegasys.teku.infrastructure.async.AsyncRunnerFactory;
 import tech.pegasys.teku.infrastructure.events.EventChannels;
@@ -171,6 +172,7 @@ public SyncService create(final EventChannels eventChannels) {
             pendingPayloadAttestations);
 
     final SyncStateTracker syncStateTracker = createSyncStateTracker(forwardSyncService);
+    eventChannels.subscribe(SlotEventsChannel.class, syncStateTracker);
 
     final HistoricalBlockSyncService historicalBlockSyncService =
         createHistoricalSyncService(syncStateTracker);
@@ -207,7 +209,13 @@ protected HistoricalBlockSyncService createHistoricalSyncService(
 
   protected SyncStateTracker createSyncStateTracker(final ForwardSync forwardSync) {
     return new SyncStateTracker(
-        asyncRunner, forwardSync, p2pNetwork, getStartupTargetPeerCount, startupTimeout, metrics);
+        asyncRunner,
+        forwardSync,
+        p2pNetwork,
+        recentChainData,
+        getStartupTargetPeerCount,
+        startupTimeout,
+        metrics);
   }
 
   protected ForwardSyncService createForwardSyncService() {

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/events/SyncStateTracker.java

@@ -16,27 +16,35 @@
 import static tech.pegasys.teku.infrastructure.logging.EventLogger.EVENT_LOG;
 
 import java.time.Duration;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 import tech.pegasys.teku.beacon.sync.forward.ForwardSync;
+import tech.pegasys.teku.ethereum.events.SlotEventsChannel;
 import tech.pegasys.teku.infrastructure.async.AsyncRunner;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.logging.EventLogger;
 import tech.pegasys.teku.infrastructure.metrics.SettableGauge;
 import tech.pegasys.teku.infrastructure.metrics.TekuMetricCategory;
 import tech.pegasys.teku.infrastructure.subscribers.Subscribers;
-import tech.pegasys.teku.networking.p2p.network.P2PNetwork;
-import tech.pegasys.teku.networking.p2p.peer.Peer;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.networking.eth2.Eth2P2PNetwork;
+import tech.pegasys.teku.networking.eth2.peers.Eth2Peer;
 import tech.pegasys.teku.service.serviceutils.Service;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoice.OptimisticHeadSubscriber;
+import tech.pegasys.teku.storage.client.RecentChainData;
 
 public class SyncStateTracker extends Service
-    implements SyncStateProvider, OptimisticHeadSubscriber {
+    implements SyncStateProvider, OptimisticHeadSubscriber, SlotEventsChannel {
   private static final Logger LOG = LogManager.getLogger();
+
   private final AsyncRunner asyncRunner;
   private final ForwardSync syncService;
-  private final P2PNetwork<? extends Peer> network;
+  private final Eth2P2PNetwork network;
+  private final RecentChainData recentChainData;
   private final Subscribers<SyncStateSubscriber> subscribers = Subscribers.create(true);
   private final EventLogger eventLogger;
   private final SettableGauge isSyncingGauge;
@@ -50,19 +58,31 @@ public class SyncStateTracker extends Service
   private long syncSubscriptionId;
   private boolean headIsOptimistic = false;
 
+  /** True once we've told the user we're behind the known chain head, so we only say it once. */
+  private boolean reportedBehindKnownChainHead = false;
+
+  /**
+   * True once we've announced a sync starting and haven't yet announced reaching sync. Forward sync
+   * can start and stop many times before it gets there - retrying, or losing every peer - and none
+   * of those is a new sync as far as the user is concerned.
+   */
+  private boolean reportedSyncStart = false;
+
   private volatile SyncState currentState;
 
   public SyncStateTracker(
       final AsyncRunner asyncRunner,
       final ForwardSync syncService,
-      final P2PNetwork<? extends Peer> network,
+      final Eth2P2PNetwork network,
+      final RecentChainData recentChainData,
       final int startupTargetPeerCount,
       final Duration startupTimeout,
       final MetricsSystem metricsSystem) {
     this(
         asyncRunner,
         syncService,
         network,
+        recentChainData,
         startupTargetPeerCount,
         startupTimeout,
         EVENT_LOG,
@@ -72,14 +92,16 @@ public SyncStateTracker(
   SyncStateTracker(
       final AsyncRunner asyncRunner,
       final ForwardSync syncService,
-      final P2PNetwork<? extends Peer> network,
+      final Eth2P2PNetwork network,
+      final RecentChainData recentChainData,
       final int startupTargetPeerCount,
       final Duration startupTimeout,
       final EventLogger eventLogger,
       final MetricsSystem metricsSystem) {
     this.asyncRunner = asyncRunner;
     this.syncService = syncService;
     this.network = network;
+    this.recentChainData = recentChainData;
     this.startupTargetPeerCount = startupTargetPeerCount;
     this.startupTimeout = startupTimeout;
     this.eventLogger = eventLogger;
@@ -128,24 +150,104 @@ public boolean unsubscribeFromSyncStateChanges(final long subscriberId) {
     return subscribers.unsubscribe(subscriberId);
   }
 
+  /**
+   * Forward sync only notifies us when it starts and stops, so re-evaluate every slot to notice our
+   * head reaching the rest of the network while no sync is active.
+   */
+  @Override
+  public synchronized void onSlot(final UInt64 slot) {
+    updateCurrentState();
+  }
+
   private void updateCurrentState() {
     final SyncState previousState = currentState;
+    boolean behindKnownChainHead = false;
     if (headIsOptimistic) {
       currentState = syncActive ? SyncState.OPTIMISTIC_SYNCING : SyncState.AWAITING_EL;
     } else if (syncActive) {
       currentState = SyncState.SYNCING;
     } else if (startingUp) {
       currentState = SyncState.START_UP;
+    } else if (isBehindKnownChainHead()) {
+      // Forward sync isn't running - it may have found no suitable peers, or stalled and be waiting
+      // to retry - but we have no reason to believe our head is the network's. Reporting IN_SYNC
+      // here would let block production and payload attribute calculation try to regenerate a state
+      // hundreds of slots ahead of our head, which is prohibitively expensive. Forward sync remains
+      // free to start again at any time and will move us back to SYNCING.
+      behindKnownChainHead = true;
+      currentState = SyncState.SYNCING;
     } else {
       currentState = SyncState.IN_SYNC;
     }
 
+    if (behindKnownChainHead && !reportedBehindKnownChainHead) {
+      reportedBehindKnownChainHead = true;
+      knownChainHeadSlot()
+          .ifPresentOrElse(
+              knownHead ->
+                  eventLogger.syncStoppedWhileBehindHead(
+                      knownHead.minusMinZero(recentChainData.getHeadSlot()).longValue()),
+              eventLogger::notInSyncWithoutPeers);
+    }
+
     if (currentState != previousState) {
+      // the catch up, whatever caused it, is announced exactly once here
+      if (previousState.isSyncing() && currentState.isInSync()) {
+        reportedBehindKnownChainHead = false;
+        reportedSyncStart = false;
+        eventLogger.syncCompleted();
+      }
       isSyncingGauge.set(currentState.isSyncing() ? 1.0 : 0.0);
       subscribers.deliver(SyncStateSubscriber::onSyncStateChange, currentState);
     }
   }
 
+  /**
+   * True when we can't believe our own head is the network's. Deliberately says nothing about the
+   * current slot: our head block being old only means we're behind if there are actually blocks out
+   * there we're missing. If the chain itself has a long run of empty slots our peers' heads are
+   * just as old as ours, so we correctly stay in sync and keep proposing.
+   *
+   * <p>Uses the same tolerance {@link RecentChainData#isCloseToInSync()} applies to enable gossip,
+   * just measured against the network's head rather than the current slot.
+   */
+  private boolean isBehindKnownChainHead() {
+    return knownChainHeadSlot()
+        .map(knownHead -> !recentChainData.isHeadCloseToSlot(knownHead))
+        // With nobody to compare against we can't tell a stale head from being at the tip. A node
+        // configured to expect peers but left with none has to assume the worst, otherwise it would
+        // try to build on a head that may be hours old. A node configured for no peers (a single
+        // node network) is authoritative on its own and must carry on proposing.
+        .orElseGet(this::expectsPeers);
+  }
+
+  private boolean expectsPeers() {
+    return startupTargetPeerCount > 0;
+  }
+
+  /**
+   * The head slot we believe the network is at, taken from peer status messages. Once we have
+   * enough peers to have a choice we take the second highest head, so that a single peer
+   * overstating its head can't convince us we're behind. Empty when we have no peers to ask.
+   */
+  private Optional<UInt64> knownChainHeadSlot() {
+    final List<UInt64> peerHeadSlots =
+        network
+            .streamPeers()
+            .filter(Eth2Peer::hasStatus)
+            .map(peer -> peer.getStatus().getHeadSlot())
+            .sorted(Comparator.reverseOrder())
+            .toList();
+    if (peerHeadSlots.isEmpty()) {
+      return Optional.empty();
+    }
+    // sorted highest first: second highest to exclude anomalies when we have the choice, otherwise
+    // the highest we have
+    final int trustedHeadIndex =
+        peerHeadSlots.size() >= Math.max(2, startupTargetPeerCount) ? 1 : 0;
+    return Optional.of(peerHeadSlots.get(trustedHeadIndex));
+  }
+
   @Override
   protected synchronized SafeFuture<?> doStart() {
     LOG.debug(
@@ -190,9 +292,8 @@ private void logSyncStateOnOptimisticHeadChanged(
 
     if (syncActive) {
       eventLogger.headNoLongerOptimisticWhileSyncing();
-    } else {
-      eventLogger.syncCompleted();
     }
+    // otherwise we're leaving AWAITING_EL, and updateCurrentState() announces the catch up
   }
 
   private void logSyncStateOnSyncingChanged(
@@ -203,15 +304,17 @@ private void logSyncStateOnSyncingChanged(
     }
 
     if (isSyncing) {
-      eventLogger.syncStart();
+      if (!reportedSyncStart) {
+        reportedSyncStart = true;
+        eventLogger.syncStart();
+      }
       return;
     }
 
     if (headIsOptimistic) {
       eventLogger.syncCompletedWhileHeadIsOptimistic();
-    } else {
-      eventLogger.syncCompleted();
     }
+    // otherwise updateCurrentState() announces it, if this stop actually reached the head
   }
 
   private synchronized void markStartupComplete() {