sync fixes (#1005)

* sync fixes

* fix Status message finalized info
* work around sync starting before initial status exchange
* don't fail block on deposit signature check failure (fixes #989)
* print ForkDigest and Version nicely
* dump incoming blocks
* fix crash when libp2p peer connection is closed
* update chunk size to 16 to work around missing blocks when syncing

* bump libp2p

* bump libp2p

* better deposit skip message
This commit is contained in:
Jacek Sieka 2020-05-11 20:08:52 +02:00 committed by GitHub
parent a7a50824a1
commit fb2e0ddbec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 148 additions and 94 deletions

View File

@ -22,7 +22,7 @@ import
spec/presets/custom, spec/presets/custom,
conf, time, beacon_chain_db, validator_pool, extras, conf, time, beacon_chain_db, validator_pool, extras,
attestation_pool, block_pool, eth2_network, eth2_discovery, attestation_pool, block_pool, eth2_network, eth2_discovery,
beacon_node_common, beacon_node_types, beacon_node_common, beacon_node_types, sszdump,
mainchain_monitor, version, ssz, ssz/dynamic_navigator, mainchain_monitor, version, ssz, ssz/dynamic_navigator,
sync_protocol, request_manager, validator_keygen, interop, statusbar, sync_protocol, request_manager, validator_keygen, interop, statusbar,
sync_manager, state_transition, sync_manager, state_transition,
@ -291,6 +291,9 @@ proc storeBlock(node: BeaconNode, signedBlock: SignedBeaconBlock): bool =
cat = "block_listener", cat = "block_listener",
pcs = "receive_block" pcs = "receive_block"
if node.config.dumpEnabled:
dump(node.config.dumpDir / "incoming", signedBlock, blockRoot)
beacon_blocks_received.inc() beacon_blocks_received.inc()
if node.blockPool.add(blockRoot, signedBlock).isNil: if node.blockPool.add(blockRoot, signedBlock).isNil:
return false return false
@ -479,35 +482,6 @@ proc onSecond(node: BeaconNode, moment: Moment) {.async.} =
discard setTimer(nextSecond) do (p: pointer): discard setTimer(nextSecond) do (p: pointer):
asyncCheck node.onSecond(nextSecond) asyncCheck node.onSecond(nextSecond)
proc getHeadSlot*(peer: Peer): Slot {.inline.} =
## Returns head slot for specific peer ``peer``.
result = peer.state(BeaconSync).statusMsg.headSlot
proc updateStatus*(peer: Peer): Future[bool] {.async.} =
## Request `status` of remote peer ``peer``.
let
nstate = peer.networkState(BeaconSync)
finalizedHead = nstate.blockPool.finalizedHead
headBlock = nstate.blockPool.head.blck
ourStatus = StatusMsg(
forkDigest: nstate.forkDigest,
finalizedRoot: finalizedHead.blck.root,
finalizedEpoch: finalizedHead.slot.compute_epoch_at_slot(),
headRoot: headBlock.root,
headSlot: headBlock.slot
)
let theirFut = awaitne peer.status(ourStatus,
timeout = chronos.seconds(60))
if theirFut.failed():
result = false
else:
let theirStatus = theirFut.read()
if theirStatus.isSome():
peer.state(BeaconSync).statusMsg = theirStatus.get()
result = true
proc runSyncLoop(node: BeaconNode) {.async.} = proc runSyncLoop(node: BeaconNode) {.async.} =
proc getLocalHeadSlot(): Slot = proc getLocalHeadSlot(): Slot =
result = node.blockPool.head.blck.slot result = node.blockPool.head.blck.slot
@ -544,7 +518,7 @@ proc runSyncLoop(node: BeaconNode) {.async.} =
node.network.peerPool, getLocalHeadSlot, getLocalWallSlot, node.network.peerPool, getLocalHeadSlot, getLocalWallSlot,
updateLocalBlocks, updateLocalBlocks,
# TODO increase when block processing perf improves # TODO increase when block processing perf improves
chunkSize = 8 chunkSize = 16
) )
await syncman.sync() await syncman.sync()

View File

@ -268,7 +268,7 @@ proc tryGetForkDigest(hexdigest: string): Option[ForkDigest] =
var res: ForkDigest var res: ForkDigest
if len(hexdigest) > 0: if len(hexdigest) > 0:
try: try:
hexToByteArray(hexdigest, res) hexToByteArray(hexdigest, array[4 ,byte](res))
result = some(res) result = some(res)
except CatchableError: except CatchableError:
discard discard
@ -441,8 +441,8 @@ proc logEnrAddress(address: string) =
if eth2Data.isSome(): if eth2Data.isSome():
try: try:
var forkid = SSZ.decode(eth2Data.get(), ENRForkID) var forkid = SSZ.decode(eth2Data.get(), ENRForkID)
eth2fork_digest = bu.toHex(forkid.fork_digest) eth2fork_digest = $forkid.fork_digest
eth2next_fork_version = bu.toHex(forkid.next_fork_version) eth2next_fork_version = $forkid.next_fork_version
eth2next_fork_epoch = strutils.toHex(cast[uint64](forkid.next_fork_epoch)) eth2next_fork_epoch = strutils.toHex(cast[uint64](forkid.next_fork_epoch))
except CatchableError: except CatchableError:
eth2fork_digest = "Error" eth2fork_digest = "Error"
@ -634,8 +634,8 @@ proc run(conf: InspectorConf) {.async.} =
if forkDigest.get() != forkOpt.get(): if forkDigest.get() != forkOpt.get():
warn "Bootstrap node address has different forkDigest", warn "Bootstrap node address has different forkDigest",
address = item.addressRec.toUri(), address = item.addressRec.toUri(),
address_fork_digest = bu.toHex(forkOpt.get()), address_fork_digest = $(forkOpt.get()),
stored_fork_digest = bu.toHex(forkDigest.get()) stored_fork_digest = $(forkDigest.get())
else: else:
forkDigest = forkOpt forkDigest = forkOpt
@ -673,8 +673,8 @@ proc run(conf: InspectorConf) {.async.} =
if argForkDigest.isSome(): if argForkDigest.isSome():
if forkDigest.isSome() != argForkDigest.isSome(): if forkDigest.isSome() != argForkDigest.isSome():
warn "forkDigest argument value is different, using argument value", warn "forkDigest argument value is different, using argument value",
argument_fork_digest = toHex(argForkDigest.get()), argument_fork_digest = argForkDigest.get(),
bootstrap_fork_digest = toHex(forkDigest.get()) bootstrap_fork_digest = forkDigest.get()
forkDigest = argForkDigest forkDigest = argForkDigest
let seckey = lcrypto.PrivateKey.random(PKScheme.Secp256k1) let seckey = lcrypto.PrivateKey.random(PKScheme.Secp256k1)

View File

@ -53,7 +53,7 @@ func makeDeposit*(
withdrawal_credentials: makeWithdrawalCredentials(pubkey))) withdrawal_credentials: makeWithdrawalCredentials(pubkey)))
if skipBLSValidation notin flags: if skipBLSValidation notin flags:
let domain = compute_domain(DOMAIN_DEPOSIT, GENESIS_FORK_VERSION) let domain = compute_domain(DOMAIN_DEPOSIT)
let signing_root = compute_signing_root(ret.getDepositMessage, domain) let signing_root = compute_signing_root(ret.getDepositMessage, domain)
ret.data.signature = bls_sign(privkey, signing_root.data) ret.data.signature = bls_sign(privkey, signing_root.data)

View File

@ -62,6 +62,9 @@ proc process_deposit*(
state.eth1_deposit_index, state.eth1_deposit_index,
state.eth1_data.deposit_root, state.eth1_data.deposit_root,
): ):
notice "Deposit merkle validation failed",
proof = deposit.proof, deposit_root = state.eth1_data.deposit_root,
deposit_index = state.eth1_deposit_index
return false return false
# Deposits must be processed in order # Deposits must be processed in order
@ -78,18 +81,17 @@ proc process_deposit*(
# by the deposit contract # by the deposit contract
# Fork-agnostic domain since deposits are valid across forks # Fork-agnostic domain since deposits are valid across forks
# let domain = compute_domain(DOMAIN_DEPOSIT)
# TODO zcli/zrnt does use the GENESIS_FORK_VERSION which can
# vary between minimal/mainnet, though, despite the comment,
# which is copied verbatim from the eth2 beacon chain spec.
# https://github.com/protolambda/zrnt/blob/v0.11.0/eth2/phase0/kickstart.go#L58
let domain = compute_domain(DOMAIN_DEPOSIT, GENESIS_FORK_VERSION)
let signing_root = compute_signing_root(deposit.getDepositMessage, domain) let signing_root = compute_signing_root(deposit.getDepositMessage, domain)
if skipBLSValidation notin flags and not bls_verify( if skipBLSValidation notin flags and not bls_verify(
pubkey, signing_root.data, pubkey, signing_root.data,
deposit.data.signature): deposit.data.signature):
return false # It's ok that deposits fail - they get included in blocks regardless
# TODO spec test?
debug "Skipping deposit with invalid signature",
pubkey, signing_root, signature = deposit.data.signature
return true
# Add validator and balance entries # Add validator and balance entries
state.validators.add(Validator( state.validators.add(Validator(
@ -220,8 +222,8 @@ proc initialize_beacon_state_from_eth1*(
const SECONDS_PER_DAY = uint64(60*60*24) const SECONDS_PER_DAY = uint64(60*60*24)
var state = BeaconStateRef( var state = BeaconStateRef(
fork: Fork( fork: Fork(
previous_version: GENESIS_FORK_VERSION, previous_version: Version(GENESIS_FORK_VERSION),
current_version: GENESIS_FORK_VERSION, current_version: Version(GENESIS_FORK_VERSION),
epoch: GENESIS_EPOCH), epoch: GENESIS_EPOCH),
genesis_time: genesis_time:
eth1_timestamp + 2'u64 * SECONDS_PER_DAY - eth1_timestamp + 2'u64 * SECONDS_PER_DAY -

View File

@ -152,8 +152,8 @@ type
data*: AttestationData data*: AttestationData
signature*: ValidatorSig signature*: ValidatorSig
Version* = array[4, byte] # TODO Maybe make this distinct Version* = distinct array[4, byte]
ForkDigest* = array[4, byte] # TODO Maybe make this distinct ForkDigest* = distinct array[4, byte]
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#forkdata # https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#forkdata
ForkData* = object ForkData* = object
@ -501,6 +501,17 @@ proc writeValue*(writer: var JsonWriter, value: ValidatorIndex) =
proc readValue*(reader: var JsonReader, value: var ValidatorIndex) = proc readValue*(reader: var JsonReader, value: var ValidatorIndex) =
value = ValidatorIndex reader.readValue(uint32) value = ValidatorIndex reader.readValue(uint32)
proc writeValue*(writer: var JsonWriter, value: Version | ForkDigest) =
writeValue(writer, $value)
proc readValue*(reader: var JsonReader, value: var Version) =
let hex = reader.readValue(string)
hexToByteArray(hex, array[4, byte](value))
proc readValue*(reader: var JsonReader, value: var ForkDigest) =
let hex = reader.readValue(string)
hexToByteArray(hex, array[4, byte](value))
# `ValidatorIndex` seq handling. # `ValidatorIndex` seq handling.
proc max*(a: ValidatorIndex, b: int) : auto = proc max*(a: ValidatorIndex, b: int) : auto =
max(a.int, b) max(a.int, b)
@ -580,6 +591,17 @@ when useListType:
else: else:
template asSeq*[T; N](x: List[T, N]): auto = x template asSeq*[T; N](x: List[T, N]): auto = x
func `$`*(v: ForkDigest | Version): string =
toHex(array[4, byte](v))
# TODO where's borrow support when you need it
func `==`*(a, b: ForkDigest | Version): bool =
array[4, byte](a) == array[4, byte](b)
func len*(v: ForkDigest | Version): int = sizeof(v)
func low*(v: ForkDigest | Version): int = 0
func high*(v: ForkDigest | Version): int = len(v) - 1
func `[]`*(v: ForkDigest | Version, idx: int): byte = array[4, byte](v)[idx]
func shortLog*(s: Slot): uint64 = func shortLog*(s: Slot): uint64 =
s - GENESIS_SLOT s - GENESIS_SLOT

View File

@ -125,7 +125,7 @@ func int_to_bytes4*(x: uint64): array[4, byte] =
result[3] = ((x shr 24) and 0xff).byte result[3] = ((x shr 24) and 0xff).byte
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_fork_data_root # https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_fork_data_root
func compute_fork_data_root(current_version: array[4, byte], func compute_fork_data_root(current_version: Version,
genesis_validators_root: Eth2Digest): Eth2Digest = genesis_validators_root: Eth2Digest): Eth2Digest =
# Return the 32-byte fork data root for the ``current_version`` and # Return the 32-byte fork data root for the ``current_version`` and
# ``genesis_validators_root``. # ``genesis_validators_root``.
@ -137,19 +137,20 @@ func compute_fork_data_root(current_version: array[4, byte],
)) ))
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_fork_digest # https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_fork_digest
func compute_fork_digest*(current_version: array[4, byte], func compute_fork_digest*(current_version: Version,
genesis_validators_root: Eth2Digest): array[4, byte] = genesis_validators_root: Eth2Digest): ForkDigest =
# Return the 4-byte fork digest for the ``current_version`` and # Return the 4-byte fork digest for the ``current_version`` and
# ``genesis_validators_root``. # ``genesis_validators_root``.
# This is a digest primarily used for domain separation on the p2p layer. # This is a digest primarily used for domain separation on the p2p layer.
# 4-bytes suffices for practical separation of forks/chains. # 4-bytes suffices for practical separation of forks/chains.
result[0..3] = array[4, byte](result)[0..3] =
compute_fork_data_root(current_version, genesis_validators_root).data[0..3] compute_fork_data_root(
current_version, genesis_validators_root).data.toOpenArray(0, 3)
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_domain # https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/beacon-chain.md#compute_domain
func compute_domain*( func compute_domain*(
domain_type: DomainType, domain_type: DomainType,
fork_version: array[4, byte] = [0'u8, 0, 0, 0], fork_version: Version = Version(GENESIS_FORK_VERSION),
genesis_validators_root: Eth2Digest = ZERO_HASH): Domain = genesis_validators_root: Eth2Digest = ZERO_HASH): Domain =
# Return the domain for the ``domain_type`` and ``fork_version``. # Return the domain for the ``domain_type`` and ``fork_version``.
let fork_data_root = let fork_data_root =

View File

@ -9,7 +9,6 @@
import import
strformat, strformat,
stew/byteutils,
datatypes datatypes
const const
@ -30,31 +29,31 @@ const
func getBeaconBlocksTopic*(forkDigest: ForkDigest): string = func getBeaconBlocksTopic*(forkDigest: ForkDigest): string =
try: try:
&"/eth2/{toHex forkDigest}/{topicBeaconBlocksSuffix}" &"/eth2/{$forkDigest}/{topicBeaconBlocksSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg
func getVoluntaryExitsTopic*(forkDigest: ForkDigest): string = func getVoluntaryExitsTopic*(forkDigest: ForkDigest): string =
try: try:
&"/eth2/{toHex forkDigest}/{topicVoluntaryExitsSuffix}" &"/eth2/{$forkDigest}/{topicVoluntaryExitsSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg
func getProposerSlashingsTopic*(forkDigest: ForkDigest): string = func getProposerSlashingsTopic*(forkDigest: ForkDigest): string =
try: try:
&"/eth2/{toHex forkDigest}/{topicProposerSlashingsSuffix}" &"/eth2/{$forkDigest}/{topicProposerSlashingsSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg
func getAttesterSlashingsTopic*(forkDigest: ForkDigest): string = func getAttesterSlashingsTopic*(forkDigest: ForkDigest): string =
try: try:
&"/eth2/{toHex forkDigest}/{topicAttesterSlashingsSuffix}" &"/eth2/{$forkDigest}/{topicAttesterSlashingsSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg
func getAggregateAndProofsTopic*(forkDigest: ForkDigest): string = func getAggregateAndProofsTopic*(forkDigest: ForkDigest): string =
try: try:
&"/eth2/{toHex forkDigest}/{topicAggregateAndProofsSuffix}" &"/eth2/{$forkDigest}/{topicAggregateAndProofsSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg
@ -62,6 +61,6 @@ func getAttestationTopic*(forkDigest: ForkDigest, committeeIndex: uint64): strin
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/validator.md#broadcast-attestation # https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/validator.md#broadcast-attestation
try: try:
let topicIndex = committeeIndex mod ATTESTATION_SUBNET_COUNT let topicIndex = committeeIndex mod ATTESTATION_SUBNET_COUNT
&"/eth2/{toHex forkDigest}/committee_index{topicIndex}{topicAttestationsSuffix}" &"/eth2/{$forkDigest}/committee_index{topicIndex}{topicAttestationsSuffix}"
except ValueError as e: except ValueError as e:
raiseAssert e.msg raiseAssert e.msg

View File

@ -13,7 +13,7 @@
#{.push raises: [Defect].} #{.push raises: [Defect].}
import import
stew/shims/macros, options, algorithm, options, stew/shims/macros, options, algorithm, options, strformat,
stew/[bitops2, bitseqs, endians2, objects, varints, ptrops, ranges/ptr_arith], stint, stew/[bitops2, bitseqs, endians2, objects, varints, ptrops, ranges/ptr_arith], stint,
faststreams/input_stream, serialization, serialization/testing/tracing, faststreams/input_stream, serialization, serialization/testing/tracing,
./spec/[crypto, datatypes, digest], ./spec/[crypto, datatypes, digest],
@ -86,7 +86,10 @@ proc mount*(F: type SSZ, stream: InputStream, T: type): T {.raises: [Defect].} =
method formatMsg*(err: ref SszSizeMismatchError, filename: string): string {.gcsafe, raises: [Defect].} = method formatMsg*(err: ref SszSizeMismatchError, filename: string): string {.gcsafe, raises: [Defect].} =
# TODO: implement proper error string # TODO: implement proper error string
"Serialisation error while processing " & filename try:
&"SSZ size mismatch, element {err.elementSize}, actual {err.actualSszSize}, type {err.deserializedType}, file {filename}"
except CatchableError:
"SSZ size mismatch"
when false: when false:
# TODO: Nim can't handle yet this simpler definition. File an issue. # TODO: Nim can't handle yet this simpler definition. File an issue.
@ -110,6 +113,7 @@ template toSszType*(x: auto): auto =
elif x is BlsCurveType: toRaw(x) elif x is BlsCurveType: toRaw(x)
elif x is BitSeq|BitList: ByteList(x) elif x is BitSeq|BitList: ByteList(x)
elif x is TypeWithMaxLen: toSszType valueOf(x) elif x is TypeWithMaxLen: toSszType valueOf(x)
elif x is ForkDigest|Version: array[4, byte](x)
elif useListType and x is List: seq[x.T](x) elif useListType and x is List: seq[x.T](x)
else: x else: x

View File

@ -51,6 +51,16 @@ template fromSszBytes*(T: type Slot, bytes: openarray[byte]): Slot =
template fromSszBytes*(T: type Epoch, bytes: openarray[byte]): Epoch = template fromSszBytes*(T: type Epoch, bytes: openarray[byte]): Epoch =
Epoch fromSszBytes(uint64, bytes) Epoch fromSszBytes(uint64, bytes)
func fromSszBytes*(T: type ForkDigest, bytes: openarray[byte]): T {.raisesssz.} =
if bytes.len < sizeof(result):
raise newException(MalformedSszError, "SSZ input of insufficient size")
copyMem(result.addr, unsafeAddr bytes[0], sizeof(result))
func fromSszBytes*(T: type Version, bytes: openarray[byte]): T {.raisesssz.} =
if bytes.len < sizeof(result):
raise newException(MalformedSszError, "SSZ input of insufficient size")
copyMem(result.addr, unsafeAddr bytes[0], sizeof(result))
template fromSszBytes*(T: type enum, bytes: openarray[byte]): auto = template fromSszBytes*(T: type enum, bytes: openarray[byte]): auto =
T fromSszBytes(uint64, bytes) T fromSszBytes(uint64, bytes)

View File

@ -8,10 +8,14 @@ import
proc dump*(dir: string, v: AttestationData, validator: ValidatorPubKey) = proc dump*(dir: string, v: AttestationData, validator: ValidatorPubKey) =
SSZ.saveFile(dir / &"att-{v.slot}-{v.index}-{shortLog(validator)}.ssz", v) SSZ.saveFile(dir / &"att-{v.slot}-{v.index}-{shortLog(validator)}.ssz", v)
proc dump*(dir: string, v: SignedBeaconBlock, root: Eth2Digest) =
SSZ.saveFile(dir / &"block-{v.message.slot}-{shortLog(root)}.ssz", v)
proc dump*(dir: string, v: SignedBeaconBlock, blck: BlockRef) = proc dump*(dir: string, v: SignedBeaconBlock, blck: BlockRef) =
SSZ.saveFile(dir / &"block-{v.message.slot}-{shortLog(blck.root)}.ssz", v) dump(dir, v, blck.root)
proc dump*(dir: string, v: HashedBeaconState, blck: BlockRef) = proc dump*(dir: string, v: HashedBeaconState, blck: BlockRef) =
SSZ.saveFile( SSZ.saveFile(
dir / &"state-{v.data.slot}-{shortLog(blck.root)}-{shortLog(v.root)}.ssz", dir / &"state-{v.data.slot}-{shortLog(blck.root)}-{shortLog(v.root)}.ssz",
v.data) v.data)

View File

@ -437,6 +437,12 @@ proc sync*[A, B](man: SyncManager[A, B]) {.async.} =
wall_head_slot = wallSlot, local_head_slot = headSlot, wall_head_slot = wallSlot, local_head_slot = headSlot,
peer_score = peer.getScore(), topics = "syncman" peer_score = peer.getScore(), topics = "syncman"
man.pool.release(peer) man.pool.release(peer)
elif not peer.hasInitialStatus:
# TODO Don't even consider these peers!
debug "Peer not ready", peer
man.pool.release(peer)
# TODO goes into tight loop without this
await sleepAsync(RESP_TIMEOUT)
else: else:
if headSlot > man.queue.lastSlot: if headSlot > man.queue.lastSlot:
man.queue = SyncQueue.init(headSlot, wallSlot, man.chunkSize, man.queue = SyncQueue.init(headSlot, wallSlot, man.chunkSize,

View File

@ -47,33 +47,41 @@ const
# request - typically clients will ask for an epoch or so at a time, but we # request - typically clients will ask for an epoch or so at a time, but we
# allow a little bit more in case they want to stream blocks faster # allow a little bit more in case they want to stream blocks faster
proc shortLog*(s: StatusMsg): auto =
(
forkDigest: s.forkDigest,
finalizedRoot: shortLog(s.finalizedRoot),
finalizedEpoch: shortLog(s.finalizedEpoch),
headRoot: shortLog(s.headRoot),
headSlot: shortLog(s.headSlot)
)
chronicles.formatIt(StatusMsg): shortLog(it)
proc importBlocks(state: BeaconSyncNetworkState, proc importBlocks(state: BeaconSyncNetworkState,
blocks: openarray[SignedBeaconBlock]) {.gcsafe.} = blocks: openarray[SignedBeaconBlock]) {.gcsafe.} =
for blk in blocks: for blk in blocks:
state.onBeaconBlock(blk) state.onBeaconBlock(blk)
info "Forward sync imported blocks", len = blocks.len info "Forward sync imported blocks", len = blocks.len
proc getCurrentStatus(state: BeaconSyncNetworkState): StatusMsg {.gcsafe.} = proc getCurrentStatus*(state: BeaconSyncNetworkState): StatusMsg {.gcsafe.} =
let let
blockPool = state.blockPool blockPool = state.blockPool
finalizedHead = blockPool.finalizedHead
headBlock = blockPool.head.blck headBlock = blockPool.head.blck
headRoot = headBlock.root
headSlot = headBlock.slot
finalizedEpoch = finalizedHead.slot.compute_epoch_at_slot()
StatusMsg( StatusMsg(
forkDigest: state.forkDigest, forkDigest: state.forkDigest,
finalizedRoot: finalizedHead.blck.root, finalizedRoot: blockPool.headState.data.data.finalized_checkpoint.root,
finalizedEpoch: finalizedEpoch, finalizedEpoch: blockPool.headState.data.data.finalized_checkpoint.epoch,
headRoot: headRoot, headRoot: headBlock.root,
headSlot: headSlot) headSlot: headBlock.slot)
proc handleInitialStatus(peer: Peer, proc handleInitialStatus(peer: Peer,
state: BeaconSyncNetworkState, state: BeaconSyncNetworkState,
ourStatus: StatusMsg, ourStatus: StatusMsg,
theirStatus: StatusMsg): Future[bool] {.async, gcsafe.} theirStatus: StatusMsg): Future[bool] {.async, gcsafe.}
proc setStatusMsg(peer: Peer, statusMsg: StatusMsg) {.gcsafe.}
p2pProtocol BeaconSync(version = 1, p2pProtocol BeaconSync(version = 1,
rlpxName = "bcs", rlpxName = "bcs",
networkState = BeaconSyncNetworkState, networkState = BeaconSyncNetworkState,
@ -92,7 +100,7 @@ p2pProtocol BeaconSync(version = 1,
let res = await peer.handleInitialStatus(peer.networkState, let res = await peer.handleInitialStatus(peer.networkState,
ourStatus, tstatus) ourStatus, tstatus)
if res: if res:
peer.state(BeaconSync).statusMsg = tstatus peer.setStatusMsg(tstatus)
else: else:
warn "Status response not received in time" warn "Status response not received in time"
@ -101,16 +109,14 @@ p2pProtocol BeaconSync(version = 1,
let let
ourStatus = peer.networkState.getCurrentStatus() ourStatus = peer.networkState.getCurrentStatus()
if not await peer.handleInitialStatus(
peer.networkState, ourStatus, theirStatus):
return
peer.setStatusMsg(theirStatus)
trace "Sending status msg", ourStatus trace "Sending status msg", ourStatus
await response.send(ourStatus) await response.send(ourStatus)
if not peer.state.initialStatusReceived:
peer.state.initialStatusReceived = true
let res = await peer.handleInitialStatus(peer.networkState,
ourStatus, theirStatus)
if res:
peer.state(BeaconSync).statusMsg = theirStatus
proc statusResp(peer: Peer, msg: StatusMsg) proc statusResp(peer: Peer, msg: StatusMsg)
proc goodbye(peer: Peer, reason: DisconnectionReason) {.libp2pProtocol("goodbye", 1).} proc goodbye(peer: Peer, reason: DisconnectionReason) {.libp2pProtocol("goodbye", 1).}
@ -179,19 +185,44 @@ p2pProtocol BeaconSync(version = 1,
peer: Peer, peer: Peer,
blocks: openarray[SignedBeaconBlock]) blocks: openarray[SignedBeaconBlock])
proc setStatusMsg(peer: Peer, statusMsg: StatusMsg) =
debug "Peer status", peer, statusMsg
peer.state(BeaconSync).initialStatusReceived = true
peer.state(BeaconSync).statusMsg = statusMsg
proc updateStatus*(peer: Peer): Future[bool] {.async.} =
## Request `status` of remote peer ``peer``.
let
nstate = peer.networkState(BeaconSync)
ourStatus = getCurrentStatus(nstate)
let theirFut = awaitne peer.status(ourStatus,
timeout = chronos.seconds(60))
if theirFut.failed():
result = false
else:
let theirStatus = theirFut.read()
if theirStatus.isSome():
peer.setStatusMsg(theirStatus.get())
result = true
proc hasInitialStatus*(peer: Peer): bool {.inline.} =
## Returns head slot for specific peer ``peer``.
peer.state(BeaconSync).initialStatusReceived
proc getHeadSlot*(peer: Peer): Slot {.inline.} =
## Returns head slot for specific peer ``peer``.
result = peer.state(BeaconSync).statusMsg.headSlot
proc handleInitialStatus(peer: Peer, proc handleInitialStatus(peer: Peer,
state: BeaconSyncNetworkState, state: BeaconSyncNetworkState,
ourStatus: StatusMsg, ourStatus: StatusMsg,
theirStatus: StatusMsg): Future[bool] {.async, gcsafe.} = theirStatus: StatusMsg): Future[bool] {.async, gcsafe.} =
if theirStatus.forkDigest != state.forkDigest: if theirStatus.forkDigest != state.forkDigest:
notice "Irrelevant peer", notice "Irrelevant peer", peer, theirStatus, ourStatus
peer, theirFork = theirStatus.forkDigest, ourFork = state.forkDigest
await peer.disconnect(IrrelevantNetwork) await peer.disconnect(IrrelevantNetwork)
return false return false
debug "Peer connected", peer, debug "Peer connected", peer, theirStatus, ourStatus
localHeadSlot = ourStatus.headSlot,
remoteHeadSlot = theirStatus.headSlot,
remoteHeadRoot = theirStatus.headRoot
return true return true
proc initBeaconSync*(network: Eth2Node, proc initBeaconSync*(network: Eth2Node,

View File

@ -116,6 +116,7 @@ cli do (skipGoerliKey {.
exec &"""nim c {nimFlags} -d:"const_preset={preset}" -o:"{beaconNodeBinary}" beacon_chain/beacon_node.nim""" exec &"""nim c {nimFlags} -d:"const_preset={preset}" -o:"{beaconNodeBinary}" beacon_chain/beacon_node.nim"""
mkDir dumpDir mkDir dumpDir
mkDir dumpDir / "incoming"
proc execIgnoringExitCode(s: string) = proc execIgnoringExitCode(s: string) =
# reduces the error output when interrupting an external command with Ctrl+C # reduces the error output when interrupting an external command with Ctrl+C

View File

@ -25,7 +25,7 @@ func signMockDepositData(
# No state --> Genesis # No state --> Genesis
let domain = compute_domain( let domain = compute_domain(
DOMAIN_DEPOSIT, DOMAIN_DEPOSIT,
default(array[4, byte]) # Genesis is fork_version 0 Version(GENESIS_FORK_VERSION)
) )
let signing_root = compute_signing_root( let signing_root = compute_signing_root(
deposit_data.getDepositMessage(), deposit_data.getDepositMessage(),
@ -43,7 +43,7 @@ func signMockDepositData(
) = ) =
let domain = compute_domain( let domain = compute_domain(
DOMAIN_DEPOSIT, DOMAIN_DEPOSIT,
default(array[4, byte]) # Genesis is fork_version 0 Version(GENESIS_FORK_VERSION)
) )
let signing_root = compute_signing_root( let signing_root = compute_signing_root(
deposit_data.getDepositMessage(), deposit_data.getDepositMessage(),

View File

@ -43,7 +43,7 @@ func makeDeposit(i: int, flags: UpdateFlags): Deposit =
privkey = makeFakeValidatorPrivKey(i) privkey = makeFakeValidatorPrivKey(i)
pubkey = privkey.toPubKey() pubkey = privkey.toPubKey()
withdrawal_credentials = makeFakeHash(i) withdrawal_credentials = makeFakeHash(i)
domain = compute_domain(DOMAIN_DEPOSIT, GENESIS_FORK_VERSION) domain = compute_domain(DOMAIN_DEPOSIT, Version(GENESIS_FORK_VERSION))
result = Deposit( result = Deposit(
data: DepositData( data: DepositData(

2
vendor/nim-libp2p vendored

@ -1 +1 @@
Subproject commit 87e1f3c61fe65f28be6e5383df81111c58b9fd76 Subproject commit c990b5d9bd0c082741de1bdc156780f9dce32647