Track ptc assignment (#7757)

* track ptc assignment

* update All-Tests-mainnet.md

* reduce registerDuties stack usage

* fix stackoverflow in PTCDuties registration by using closure iterators

* fix: address reviews

Author: Tomi-3-0

AllTests-mainnet.md

@@ -1166,9 +1166,11 @@ AllTests-mainnet
 ```
 ## subnet tracker
 ```diff
++ should register and prune PTC duties                                                       OK
 + should register stability subnets on attester duties                                       OK
 + should register sync committee duties                                                      OK
 + should subscribe to all subnets when flag is enabled                                       OK
++ should track PTC duties in slot bitmaps                                                    OK
 ```
 ## test_fixture_ssz_generic_types.nim
 ```diff

beacon_chain/spec/beaconstate.nim

@@ -2763,8 +2763,8 @@ func can_advance_slots*(
   withState(state): forkyState.can_advance_slots(block_root, target_slot)
 
 # https://github.com/ethereum/consensus-specs/blob/v1.6.0-alpha.6/specs/gloas/beacon-chain.md#new-get_ptc
-iterator get_ptc(state: gloas.BeaconState, slot: Slot, cache: var StateCache):
-    ValidatorIndex =
+iterator get_ptc*(state: gloas.BeaconState, slot: Slot, cache: var StateCache):
+    ValidatorIndex {.closure.} =
   ## Get the payload timeliness committee for the given ``slot``
   let epoch = slot.epoch()
   var buffer {.noinit.}: array[40, byte]

beacon_chain/validators/action_tracker.nim

@@ -38,6 +38,10 @@ type
     subnet_id*: SubnetId
     slot*: Slot
 
+  PTCDuty* = object
+    slot*: Slot
+    validator_index: ValidatorIndex
+
   ActionTracker* = object
     nodeId: UInt256
 
@@ -73,6 +77,9 @@ type
     lastSyncUpdate*: Opt[SyncCommitteePeriod]
     syncDuties*: Table[ValidatorPubKey, Epoch]
 
+    ptcSlots*: array[2, uint32]
+    ptcDuties*: HashSet[PTCDuty]
+
 func hash*(x: AggregatorDuty): Hash =
   hashAllFields(x)
 
@@ -116,6 +123,32 @@ func hasSyncDuty*(
     tracker: ActionTracker, pubkey: ValidatorPubKey, epoch: Epoch): bool =
   epoch < tracker.syncDuties.getOrDefault(pubkey, GENESIS_EPOCH)
 
+proc registerPTCDuty*(
+    tracker: var ActionTracker, slot: Slot, vidx: ValidatorIndex) =
+  if slot < tracker.currentSlot or
+      slot >= Slot(uint64(tracker.currentSlot) + (SLOTS_PER_EPOCH * 2)):
+    debug "Irrelevant PTC duty", slot, vidx
+    return
+
+  tracker.knownValidators[vidx] = slot
+
+  let newDuty = PTCDuty(slot: slot, validator_index: vidx)
+
+  debug "Registering PTC duty", slot, vidx
+  tracker.ptcDuties.incl(newDuty)
+
+from std/sequtils import anyIt, toSeq
+
+func hasPTCDuty*(tracker: ActionTracker, slot: Slot): bool =
+  tracker.ptcDuties.anyIt(it.slot == slot)
+
+func getPTCDuties*(tracker: ActionTracker, slot: Slot): seq[ValidatorIndex] =
+  var duties: seq[ValidatorIndex]
+  for duty in tracker.ptcDuties:
+    if duty.slot == slot:
+      duties.add(duty.validator_index)
+  duties
+
 func aggregateSubnets*(tracker: ActionTracker, wallSlot: Slot): AttnetBits =
   var res: AttnetBits
   # Subscribe to subnets for upcoming duties
@@ -146,6 +179,7 @@ proc updateSlot*(tracker: var ActionTracker, wallSlot: Slot) =
   # are only so many slot/subnet combos - prune both internal and API-supplied
   # duties at the same time
   tracker.duties.keepItIf(it.slot >= wallSlot)
+  tracker.ptcDuties.keepItIf(it.slot >= wallSlot)
 
   block:
     var dels: seq[ValidatorPubKey]
@@ -211,8 +245,6 @@ func needsUpdate*(
   tracker.attesterDepRoot !=
     state.dependent_root(if epoch > Epoch(0): epoch - 1 else: epoch)
 
-from std/sequtils import toSeq
-
 func updateActions*(
     tracker: var ActionTracker, shufflingRef: ShufflingRef,
     beaconProposers: openArray[Opt[ValidatorIndex]]) =
@@ -232,6 +264,7 @@ func updateActions*(
         tracker.proposingSlots[epoch mod 2] or (1'u32 shl i)
 
   tracker.attestingSlots[epoch mod 2] = 0
+  tracker.ptcSlots[epoch mod 2] = 0
 
   # The relevant bitmaps are 32 bits each.
   static: doAssert SLOTS_PER_EPOCH <= 32
@@ -257,6 +290,12 @@ func updateActions*(
       tracker.attestingSlots[epoch mod 2] or
         (1'u32 shl (slot mod SLOTS_PER_EPOCH))
 
+  for duty in tracker.ptcDuties:
+    if duty.slot.epoch == epoch:
+      tracker.ptcSlots[epoch mod 2] =
+        tracker.ptcSlots[epoch mod 2] or
+          (1'u32 shl (duty.slot mod SLOTS_PER_EPOCH))
+
 func init*(
     T: type ActionTracker, nodeId: UInt256, subscribeAllAttnets: bool): T =
   T(

beacon_chain/validators/beacon_validators.nim

@@ -43,7 +43,7 @@ import
     validator_pool,
   ]
 
-from std/sequtils import mapIt
+from std/sequtils import mapIt, toSeq
 from eth/async_utils import awaitWithTimeout
 from ./message_router_mev import unblindAndRouteBlockMEV
 
@@ -1282,6 +1282,30 @@ proc handleValidatorDuties*(node: BeaconNode, lastSlot, slot: Slot) {.async: (ra
   sendAggregatedAttestations(node, head, slot)
   sendSyncCommitteeContributions(node, head, slot)
 
+proc registerPTCDuties(node: BeaconNode, epoch: Epoch) =
+  if node.dag.cfg.consensusForkAtEpoch(epoch) < ConsensusFork.Gloas:
+    return
+
+  let validatorIndices = block:
+    var res: HashSet[ValidatorIndex]
+    for idx in node.attachedValidators[].indices():
+      res.incl(idx)
+    res
+
+  withState(node.dag.headState):
+    when consensusFork >= ConsensusFork.Gloas:
+      var cache: StateCache
+      
+      for slot in epoch.slots():
+        for validator_index in get_ptc(forkyState.data, slot, cache):
+          if validator_index in validatorIndices:
+            node.consensusManager[].actionTracker.registerPTCDuty(
+              slot, validator_index)
+            
+            debug "PTC duty registered",
+              slot = slot,
+              epoch = epoch
+
 proc registerDuties*(node: BeaconNode, wallSlot: Slot) {.async: (raises: [CancelledError]).} =
   ## Register upcoming duties of attached validators with the duty tracker
 
@@ -1327,3 +1351,6 @@ proc registerDuties*(node: BeaconNode, wallSlot: Slot) {.async: (raises: [Cancel
 
         node.consensusManager[].actionTracker.registerDuty(
           slot, subnet_id, validator_index, isAggregator)
+
+  if wallSlot == wallSlot.epoch.start_slot():
+    node.registerPTCDuties(wallSlot.epoch + 1)

tests/test_action_tracker.nim

@@ -12,6 +12,9 @@ import
   unittest2,
   ../beacon_chain/validators/action_tracker
 
+from ../beacon_chain/consensus_object_pools/block_pools_types import
+  ShufflingRef
+
 suite "subnet tracker":
   test "should register stability subnets on attester duties":
     var tracker = ActionTracker.init(default(UInt256), false)
@@ -98,3 +101,60 @@ suite "subnet tracker":
     check:
       tracker.stabilitySubnets(Slot(0)).countOnes() == 64  # All 64 subnets
       tracker.aggregateSubnets(Slot(0)).countOnes() == 0
+
+  test "should register and prune PTC duties":
+    var tracker = ActionTracker.init(default(UInt256), false)
+    tracker.updateSlot(Slot(100))
+
+    check:
+      not tracker.hasPTCDuty(Slot(100))
+
+    # Register past duty
+    tracker.registerPTCDuty(Slot(99), ValidatorIndex(0))
+    check not tracker.hasPTCDuty(Slot(99))
+
+    # Register duty too far in future
+    tracker.registerPTCDuty(Slot(100 + SLOTS_PER_EPOCH * 2), ValidatorIndex(0))
+    check not tracker.hasPTCDuty(Slot(100 + SLOTS_PER_EPOCH * 2))
+
+    tracker.registerPTCDuty(Slot(105), ValidatorIndex(100))
+    tracker.registerPTCDuty(Slot(105), ValidatorIndex(101))
+    tracker.registerPTCDuty(Slot(110), ValidatorIndex(102))
+
+    check:
+      tracker.hasPTCDuty(Slot(105))
+      tracker.hasPTCDuty(Slot(110))
+      not tracker.hasPTCDuty(Slot(107))
+      tracker.knownValidators.len() == 3
+
+    # Update slot to prune old duties
+    tracker.updateSlot(Slot(107))
+    check:
+      not tracker.hasPTCDuty(Slot(105))
+      tracker.hasPTCDuty(Slot(110))
+
+    # Validator decays after a long time
+    tracker.updateSlot(Slot(110 + KNOWN_VALIDATOR_DECAY + 1))
+    check tracker.knownValidators.len() == 0
+
+  test "should track PTC duties in slot bitmaps":
+    var
+      tracker = ActionTracker.init(default(UInt256), false)
+      shufflingRef = ShufflingRef(
+        epoch: Epoch(1),
+        attester_dependent_root: ZERO_HASH,
+        shuffled_active_validator_indices: @[]
+      )
+      beaconProposers: array[SLOTS_PER_EPOCH, Opt[ValidatorIndex]]
+
+    tracker.registerPTCDuty(Slot(32), ValidatorIndex(0))  # First slot of epoch
+    tracker.registerPTCDuty(Slot(47), ValidatorIndex(1))  # Mid epoch
+    tracker.registerPTCDuty(Slot(63), ValidatorIndex(2))  # Last slot of epoch
+
+    # Update actions to populate bitmaps
+    tracker.updateActions(shufflingRef, beaconProposers)
+
+    check:
+      (tracker.ptcSlots[1] and (1'u32 shl 0)) != 0   # Slot 32
+      (tracker.ptcSlots[1] and (1'u32 shl 15)) != 0   # Slot 47
+      (tracker.ptcSlots[1] and (1'u32 shl 31)) != 0  # Slot 63