365 lines
14 KiB
Nim
365 lines
14 KiB
Nim
# beacon_chain
|
|
# Copyright (c) 2018-2020 Status Research & Development GmbH
|
|
# Licensed and distributed under either of
|
|
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
|
|
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
|
|
# at your option. This file may not be copied, modified, or distributed except according to those terms.
|
|
|
|
{.push raises: [Defect].}
|
|
|
|
import
|
|
# Standard libraries
|
|
std/[algorithm, deques, sequtils, sets, tables, options],
|
|
# Status libraries
|
|
chronicles, stew/[byteutils], json_serialization/std/sets as jsonSets,
|
|
# Internal
|
|
./spec/[beaconstate, datatypes, crypto, digest, helpers],
|
|
ssz/merkleization,
|
|
./block_pools/[spec_cache, chain_dag, clearance], ./beacon_node_types,
|
|
./fork_choice/fork_choice
|
|
|
|
export beacon_node_types, sets
|
|
|
|
logScope: topics = "attpool"
|
|
|
|
proc init*(T: type AttestationPool, chainDag: ChainDAGRef, quarantine: QuarantineRef): T =
|
|
## Initialize an AttestationPool from the chainDag `headState`
|
|
## The `finalized_root` works around the finalized_checkpoint of the genesis block
|
|
## holding a zero_root.
|
|
|
|
let
|
|
finalizedEpochRef = chainDag.getEpochRef(
|
|
chainDag.finalizedHead.blck, chainDag.finalizedHead.slot.epoch())
|
|
|
|
var forkChoice = ForkChoice.init(
|
|
finalizedEpochRef, chainDag.finalizedHead.blck)
|
|
|
|
# Feed fork choice with unfinalized history - during startup, block pool only
|
|
# keeps track of a single history so we just need to follow it
|
|
doAssert chainDag.heads.len == 1, "Init only supports a single history"
|
|
|
|
var blocks: seq[BlockRef]
|
|
var cur = chainDag.head
|
|
while cur != chainDag.finalizedHead.blck:
|
|
blocks.add cur
|
|
cur = cur.parent
|
|
|
|
debug "Preloading fork choice with blocks", blocks = blocks.len
|
|
|
|
for blck in reversed(blocks):
|
|
let
|
|
epochRef = chainDag.getEpochRef(blck, blck.slot.compute_epoch_at_slot)
|
|
status =
|
|
forkChoice.process_block(
|
|
chainDag, epochRef, blck, chainDag.get(blck).data.message, blck.slot)
|
|
|
|
doAssert status.isOk(), "Error in preloading the fork choice: " & $status.error
|
|
|
|
debug "Fork choice initialized",
|
|
justified_epoch = chainDag.headState.data.data.current_justified_checkpoint.epoch,
|
|
finalized_epoch = chainDag.headState.data.data.finalized_checkpoint.epoch,
|
|
finalized_root = shortlog(chainDag.finalizedHead.blck.root)
|
|
|
|
T(
|
|
chainDag: chainDag,
|
|
quarantine: quarantine,
|
|
forkChoice: forkChoice
|
|
)
|
|
|
|
proc addForkChoiceVotes(
|
|
pool: var AttestationPool, slot: Slot, participants: HashSet[ValidatorIndex],
|
|
block_root: Eth2Digest, wallSlot: Slot) =
|
|
# Add attestation votes to fork choice
|
|
if (let v = pool.forkChoice.on_attestation(
|
|
pool.chainDag, slot, block_root, participants, wallSlot);
|
|
v.isErr):
|
|
# This indicates that the fork choice and the chain dag are out of sync -
|
|
# this is most likely the result of a bug, but we'll try to keep going -
|
|
# hopefully the fork choice will heal itself over time.
|
|
error "Couldn't add attestation to fork choice, bug?", err = v.error()
|
|
|
|
func candidateIdx(pool: AttestationPool, slot: Slot): Option[uint64] =
|
|
if slot >= pool.startingSlot and
|
|
slot < (pool.startingSlot + pool.candidates.lenu64):
|
|
some(slot mod pool.candidates.lenu64)
|
|
else:
|
|
none(uint64)
|
|
|
|
proc updateCurrent(pool: var AttestationPool, wallSlot: Slot) =
|
|
if wallSlot + 1 < pool.candidates.lenu64:
|
|
return
|
|
|
|
if pool.startingSlot + pool.candidates.lenu64 - 1 > wallSlot:
|
|
error "Current slot older than attestation pool view, clock reset?",
|
|
poolSlot = pool.startingSlot, wallSlot
|
|
return
|
|
|
|
# As time passes we'll clear out any old attestations as they are no longer
|
|
# viable to be included in blocks
|
|
|
|
let newWallSlot = wallSlot + 1 - pool.candidates.lenu64
|
|
for i in pool.startingSlot..newWallSlot:
|
|
pool.candidates[i.uint64 mod pool.candidates.lenu64] = AttestationsSeen()
|
|
|
|
pool.startingSlot = newWallSlot
|
|
|
|
# now also clear old aggregated attestations
|
|
var keysToRemove: seq[Slot] = @[]
|
|
for k, v in pool.attestationAggregates.pairs:
|
|
if k < pool.startingSlot:
|
|
keysToRemove.add k
|
|
for k in keysToRemove:
|
|
pool.attestationAggregates.del k
|
|
|
|
proc addToAggregates(pool: var AttestationPool, attestation: Attestation) =
|
|
# do a lookup for the current slot and get it's associated htrs/attestations
|
|
var aggreated_attestation = pool.attestationAggregates.mgetOrPut(
|
|
attestation.data.slot, Table[Eth2Digest, Attestation]()).
|
|
# do a lookup for the same attestation data htr and get the attestation
|
|
mgetOrPut(attestation.data.hash_tree_root, attestation)
|
|
# if the aggregation bits differ (we didn't just insert it into the table)
|
|
# and only if there is no overlap of the signatures ==> aggregate!
|
|
if not aggreated_attestation.aggregation_bits.overlaps(attestation.aggregation_bits):
|
|
var agg {.noInit.}: AggregateSignature
|
|
agg.init(aggreated_attestation.signature)
|
|
aggreated_attestation.aggregation_bits.combine(attestation.aggregation_bits)
|
|
agg.aggregate(attestation.signature)
|
|
aggreated_attestation.signature = agg.finish()
|
|
|
|
proc addAttestation*(pool: var AttestationPool,
|
|
attestation: Attestation,
|
|
participants: HashSet[ValidatorIndex],
|
|
wallSlot: Slot) =
|
|
## Add an attestation to the pool, assuming it's been validated already.
|
|
## Attestations may be either agggregated or not - we're pursuing an eager
|
|
## strategy where we'll drop validations we already knew about and combine
|
|
## the new attestation with an existing one if possible.
|
|
##
|
|
## This strategy is not optimal in the sense that it would be possible to find
|
|
## a better packing of attestations by delaying the aggregation, but because
|
|
## it's possible to include more than one aggregate in a block we can be
|
|
## somewhat lazy instead of looking for a perfect packing.
|
|
logScope:
|
|
attestation = shortLog(attestation)
|
|
|
|
updateCurrent(pool, wallSlot)
|
|
|
|
let candidateIdx = pool.candidateIdx(attestation.data.slot)
|
|
if candidateIdx.isNone:
|
|
debug "Skipping old attestation for block production",
|
|
startingSlot = pool.startingSlot
|
|
return
|
|
|
|
pool.addToAggregates(attestation)
|
|
|
|
let
|
|
attestationsSeen = addr pool.candidates[candidateIdx.get]
|
|
validation = Validation(
|
|
aggregation_bits: attestation.aggregation_bits,
|
|
aggregate_signature: attestation.signature)
|
|
|
|
var found = false
|
|
for a in attestationsSeen.attestations.mitems():
|
|
if a.data == attestation.data:
|
|
for v in a.validations:
|
|
if validation.aggregation_bits.isSubsetOf(v.aggregation_bits):
|
|
# The validations in the new attestation are a subset of one of the
|
|
# attestations that we already have on file - no need to add this
|
|
# attestation to the database
|
|
trace "Ignoring subset attestation", newParticipants = participants
|
|
found = true
|
|
break
|
|
|
|
if not found:
|
|
# Attestations in the pool that are a subset of the new attestation
|
|
# can now be removed per same logic as above
|
|
|
|
trace "Removing subset attestations", newParticipants = participants
|
|
|
|
a.validations.keepItIf(
|
|
not it.aggregation_bits.isSubsetOf(validation.aggregation_bits))
|
|
|
|
a.validations.add(validation)
|
|
pool.addForkChoiceVotes(
|
|
attestation.data.slot, participants, attestation.data.beacon_block_root,
|
|
wallSlot)
|
|
|
|
debug "Attestation resolved",
|
|
attestation = shortLog(attestation),
|
|
validations = a.validations.len()
|
|
|
|
found = true
|
|
|
|
break
|
|
|
|
if not found:
|
|
attestationsSeen.attestations.add(AttestationEntry(
|
|
data: attestation.data,
|
|
validations: @[validation]
|
|
))
|
|
pool.addForkChoiceVotes(
|
|
attestation.data.slot, participants, attestation.data.beacon_block_root,
|
|
wallSlot)
|
|
|
|
debug "Attestation resolved",
|
|
attestation = shortLog(attestation),
|
|
validations = 1
|
|
|
|
proc addForkChoice*(pool: var AttestationPool,
|
|
epochRef: EpochRef,
|
|
blckRef: BlockRef,
|
|
blck: BeaconBlock,
|
|
wallSlot: Slot) =
|
|
## Add a verified block to the fork choice context
|
|
let state = pool.forkChoice.process_block(
|
|
pool.chainDag, epochRef, blckRef, blck, wallSlot)
|
|
|
|
if state.isErr:
|
|
# This indicates that the fork choice and the chain dag are out of sync -
|
|
# this is most likely the result of a bug, but we'll try to keep going -
|
|
# hopefully the fork choice will heal itself over time.
|
|
error "Couldn't add block to fork choice, bug?",
|
|
blck = shortLog(blck), err = state.error
|
|
|
|
proc getAttestationsForSlot*(pool: AttestationPool, newBlockSlot: Slot):
|
|
Option[AttestationsSeen] =
|
|
if newBlockSlot < (GENESIS_SLOT + MIN_ATTESTATION_INCLUSION_DELAY):
|
|
debug "Too early for attestations", newBlockSlot = shortLog(newBlockSlot)
|
|
return none(AttestationsSeen)
|
|
|
|
let
|
|
attestationSlot = newBlockSlot - MIN_ATTESTATION_INCLUSION_DELAY
|
|
candidateIdx = pool.candidateIdx(attestationSlot)
|
|
|
|
if candidateIdx.isNone:
|
|
trace "No attestations matching the slot range",
|
|
attestationSlot = shortLog(attestationSlot),
|
|
startingSlot = shortLog(pool.startingSlot)
|
|
return none(AttestationsSeen)
|
|
|
|
some(pool.candidates[candidateIdx.get()])
|
|
|
|
proc getAttestationsForBlock*(pool: AttestationPool,
|
|
state: BeaconState,
|
|
cache: var StateCache): seq[Attestation] =
|
|
## Retrieve attestations that may be added to a new block at the slot of the
|
|
## given state
|
|
let newBlockSlot = state.slot
|
|
var attestations: seq[AttestationEntry]
|
|
|
|
# This potentially creates problems with lots of repeated attestations,
|
|
# as a bunch of synchronized beacon_nodes do almost the opposite of the
|
|
# intended thing -- sure, _blocks_ have to be popular (via attestation)
|
|
# but _attestations_ shouldn't have to be so frequently repeated, as an
|
|
# artifact of this state-free, identical-across-clones choice basis. In
|
|
# addAttestation, too, the new attestations get added to the end, while in
|
|
# these functions, it's reading from the beginning, et cetera. This all
|
|
# needs a single unified strategy.
|
|
for i in max(1, newBlockSlot.int64 - ATTESTATION_LOOKBACK.int64) .. newBlockSlot.int64:
|
|
let maybeSlotData = getAttestationsForSlot(pool, i.Slot)
|
|
if maybeSlotData.isSome:
|
|
insert(attestations, maybeSlotData.get.attestations)
|
|
|
|
if attestations.len == 0:
|
|
return
|
|
|
|
for a in attestations:
|
|
var
|
|
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.3/specs/phase0/validator.md#construct-attestation
|
|
attestation = Attestation(
|
|
aggregation_bits: a.validations[0].aggregation_bits,
|
|
data: a.data,
|
|
signature: a.validations[0].aggregate_signature
|
|
)
|
|
|
|
agg {.noInit.}: AggregateSignature
|
|
agg.init(a.validations[0].aggregate_signature)
|
|
|
|
# Signature verification here is more of a sanity check - it could
|
|
# be optimized away, though for now it's easier to reuse the logic from
|
|
# the state transition function to ensure that the new block will not
|
|
# fail application.
|
|
if (let v = check_attestation(state, attestation, {}, cache); v.isErr):
|
|
warn "Attestation no longer validates...",
|
|
attestation = shortLog(attestation),
|
|
err = v.error
|
|
|
|
continue
|
|
|
|
for i in 1..a.validations.high:
|
|
if not attestation.aggregation_bits.overlaps(
|
|
a.validations[i].aggregation_bits):
|
|
attestation.aggregation_bits.combine(a.validations[i].aggregation_bits)
|
|
agg.aggregate(a.validations[i].aggregate_signature)
|
|
|
|
attestation.signature = agg.finish()
|
|
result.add(attestation)
|
|
|
|
if result.lenu64 >= MAX_ATTESTATIONS:
|
|
debug "getAttestationsForBlock: returning early after hitting MAX_ATTESTATIONS",
|
|
attestationSlot = newBlockSlot - 1
|
|
return
|
|
|
|
proc getAggregatedAttestation*(pool: AttestationPool,
|
|
slot: Slot,
|
|
ad_htr: Eth2Digest): Option[Attestation] =
|
|
try:
|
|
if pool.attestationAggregates.contains(slot) and
|
|
pool.attestationAggregates[slot].contains(ad_htr):
|
|
return some pool.attestationAggregates[slot][ad_htr]
|
|
except KeyError:
|
|
doAssert(false) # shouldn't be possible because we check with `contains`
|
|
return none(Attestation)
|
|
|
|
proc getAggregatedAttestation*(pool: AttestationPool,
|
|
slot: Slot,
|
|
index: CommitteeIndex): Option[Attestation] =
|
|
let attestations = pool.getAttestationsForSlot(
|
|
slot + MIN_ATTESTATION_INCLUSION_DELAY)
|
|
if attestations.isNone:
|
|
return none(Attestation)
|
|
|
|
for a in attestations.get.attestations:
|
|
doAssert a.data.slot == slot
|
|
if index.uint64 != a.data.index:
|
|
continue
|
|
|
|
var
|
|
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.3/specs/phase0/validator.md#construct-attestation
|
|
attestation = Attestation(
|
|
aggregation_bits: a.validations[0].aggregation_bits,
|
|
data: a.data,
|
|
signature: a.validations[0].aggregate_signature
|
|
)
|
|
|
|
agg {.noInit.}: AggregateSignature
|
|
|
|
agg.init(a.validations[0].aggregate_signature)
|
|
for v in a.validations[1..^1]:
|
|
if not attestation.aggregation_bits.overlaps(v.aggregation_bits):
|
|
attestation.aggregation_bits.combine(v.aggregation_bits)
|
|
agg.aggregate(v.aggregate_signature)
|
|
|
|
attestation.signature = agg.finish()
|
|
|
|
return some(attestation)
|
|
|
|
none(Attestation)
|
|
|
|
proc selectHead*(pool: var AttestationPool, wallSlot: Slot): BlockRef =
|
|
## Trigger fork choice and returns the new head block.
|
|
## Can return `nil`
|
|
let newHead = pool.forkChoice.get_head(pool.chainDag, wallSlot)
|
|
|
|
if newHead.isErr:
|
|
error "Couldn't select head", err = newHead.error
|
|
nil
|
|
else:
|
|
pool.chainDag.getRef(newHead.get())
|
|
|
|
proc prune*(pool: var AttestationPool) =
|
|
if (let v = pool.forkChoice.prune(); v.isErr):
|
|
# If pruning fails, it's likely the result of a bug - this shouldn't happen
|
|
# but we'll keep running hoping that the fork chocie will recover eventually
|
|
error "Couldn't prune fork choice, bug?", err = v.error()
|