vc: fix missing attestations due to doppelganger in epoch 1 (#4688)

* log validator that triggers doppelganger
* move activity detection closer to where it's performed, record
aggregation as activity (since it's part of the liveness endpoint)
* vc: check for doppelgangers in epoch 0 also (so that activity in epoch
1 can happen)
* avoid some looping when processing activities

beacon_chain/nimbus_validator_client.nim

@@ -332,7 +332,7 @@ proc asyncRun*(vc: ValidatorClientRef) {.async.} =
     vc.keymanagerServer.router.installKeymanagerHandlers(vc.keymanagerHost[])
     vc.keymanagerServer.start()
 
-  var doppelEventFut = vc.doppelExit.wait()
+  let doppelEventFut = vc.doppelExit.wait()
   try:
     vc.runSlotLoopFut = runSlotLoop(vc, vc.beaconClock.now(), onSlotStart)
     vc.runKeystoreCachePruningLoopFut =
@@ -349,7 +349,7 @@ proc asyncRun*(vc: ValidatorClientRef) {.async.} =
   await vc.shutdownMetrics()
   vc.shutdownSlashingProtection()
 
-  if doppelEventFut.finished:
+  if doppelEventFut.completed():
     # Critically, database has been shut down - the rest doesn't matter, we need
     # to stop as soon as possible
     # TODO we need to actually quit _before_ any other async tasks have had the

beacon_chain/validator_client/attestation_service.nim

@@ -26,7 +26,7 @@ proc serveAttestation(service: AttestationServiceRef, adata: AttestationData,
                       duty: DutyAndProof): Future[bool] {.async.} =
   let vc = service.client
   let validator = vc.getValidatorForDuties(
-      duty.data.pubkey, adata.slot, true).valueOr:
+      duty.data.pubkey, adata.slot).valueOr:
     return false
   let fork = vc.forkAtEpoch(adata.slot.epoch)
 
@@ -78,6 +78,8 @@ proc serveAttestation(service: AttestationServiceRef, adata: AttestationData,
         validator = shortLog(validator), validator_index = vindex,
         delay = vc.getDelay(adata.slot.attestation_deadline())
 
+  validator.doppelgangerActivity(attestation.data.slot.epoch)
+
   let res =
     try:
       await vc.submitPoolAttestations(@[attestation], ApiStrategyKind.First)
@@ -158,6 +160,8 @@ proc serveAggregateAndProof*(service: AttestationServiceRef,
         validator = shortLog(validator), validator_index = vindex,
         delay = vc.getDelay(slot.aggregate_deadline())
 
+  validator.doppelgangerActivity(proof.aggregate.data.slot.epoch)
+
   let res =
     try:
       await vc.publishAggregateAndProofs(@[signedProof], ApiStrategyKind.First)

beacon_chain/validator_client/common.nim

@@ -550,10 +550,8 @@ proc getDelay*(vc: ValidatorClientRef, deadline: BeaconTime): TimeDiff =
 
 proc getValidatorForDuties*(vc: ValidatorClientRef,
                             key: ValidatorPubKey, slot: Slot,
-                            doppelActivity = false,
                             slashingSafe = false): Opt[AttachedValidator] =
-  vc.attachedValidators[].getValidatorForDuties(
-    key, slot, doppelActivity, slashingSafe)
+  vc.attachedValidators[].getValidatorForDuties(key, slot, slashingSafe)
 
 proc forkAtEpoch*(vc: ValidatorClientRef, epoch: Epoch): Fork =
   # If schedule is present, it MUST not be empty.

beacon_chain/validator_client/doppelganger_service.nim

@@ -40,9 +40,14 @@ proc processActivities(service: DoppelgangerServiceRef, epoch: Epoch,
           validator.doppelgangerChecked(epoch)
 
           if item.is_live and validator.triggersDoppelganger(epoch):
+            warn "Doppelganger detection triggered",
+              validator = shortLog(validator), epoch
+
             vc.doppelExit.fire()
             return
 
+          break
+
 proc mainLoop(service: DoppelgangerServiceRef) {.async.} =
   let vc = service.client
   service.state = ServiceState.Running
@@ -55,8 +60,11 @@ proc mainLoop(service: DoppelgangerServiceRef) {.async.} =
 
   # On (re)start, we skip the remainder of the epoch before we start monitoring
   # for doppelgangers so we don't trigger on the attestations we produced before
-  # the epoch
-  await service.waitForNextEpoch()
+  # the epoch - there's no activity in the genesis slot, so if we start at or
+  # before that, we can safely perform the check for epoch 0 and thus keep
+  # validating in epoch 1
+  if vc.beaconClock.now().slotOrZero() > GENESIS_SLOT:
+    await service.waitForNextEpoch()
 
   while try:
     # Wait for the epoch to end - at the end (or really, the beginning of the

beacon_chain/validators/validator_duties.nim

@@ -129,13 +129,12 @@ proc getValidator*(node: BeaconNode, idx: ValidatorIndex): Opt[AttachedValidator
   node.attachedValidators[].getValidator(key.toPubKey())
 
 proc getValidatorForDuties*(
-    node: BeaconNode,
-    idx: ValidatorIndex, slot: Slot,
-    doppelActivity = false, slashingSafe = false): Opt[AttachedValidator] =
+    node: BeaconNode, idx: ValidatorIndex, slot: Slot,
+    slashingSafe = false): Opt[AttachedValidator] =
   let key = ? node.dag.validatorKey(idx)
 
   node.attachedValidators[].getValidatorForDuties(
-    key.toPubKey(), slot, doppelActivity, slashingSafe)
+    key.toPubKey(), slot, slashingSafe)
 
 proc isSynced*(node: BeaconNode, head: BlockRef): SyncStatus =
   ## TODO This function is here as a placeholder for some better heurestics to
@@ -242,6 +241,8 @@ proc createAndSendAttestation(node: BeaconNode,
           [uint64 indexInCommittee], committeeLen, data, signature).expect(
             "valid data")
 
+    validator.doppelgangerActivity(attestation.data.slot.epoch)
+
     # Logged in the router
     let res = await node.router.routeAttestation(
       attestation, subnet_id, checkSignature = false)
@@ -1136,8 +1137,7 @@ proc handleAttestations(node: BeaconNode, head: BlockRef, slot: Slot) =
       epochRef.shufflingRef, slot, committee_index)
 
     for index_in_committee, validator_index in committee:
-      let validator = node.getValidatorForDuties(
-          validator_index, slot, true).valueOr:
+      let validator = node.getValidatorForDuties(validator_index, slot).valueOr:
         continue
 
       let
@@ -1347,6 +1347,8 @@ proc signAndSendAggregate(
         return
       res.get()
 
+    validator.doppelgangerActivity(msg.message.aggregate.data.slot.epoch)
+
     # Logged in the router
     discard await node.router.routeSignedAggregateAndProof(
       msg, checkSignature = false)

beacon_chain/validators/validator_pool.nim

@@ -326,7 +326,7 @@ proc doppelgangerReady*(validator: AttachedValidator, slot: Slot): bool =
 
 proc getValidatorForDuties*(
     pool: ValidatorPool, key: ValidatorPubKey, slot: Slot,
-    doppelActivity: bool, slashingSafe: bool):
+    slashingSafe: bool):
     Opt[AttachedValidator] =
   ## Return validator only if it is ready for duties (has index and has passed
   ## doppelganger check where applicable)
@@ -348,11 +348,6 @@ proc getValidatorForDuties*(
 
     return Opt.none(AttachedValidator)
 
-  if doppelActivity:
-    # Record the activity
-    # TODO consider moving to the the "registration point"
-    validator.doppelgangerActivity(slot.epoch)
-
   return Opt.some(validator)
 
 func triggersDoppelganger*(