status-im
/
nimbus-eth2
mirror of https://github.com/status-im/nimbus-eth2.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity
nimbus-eth2/beacon_chain/mainchain_monitor.nim

817 lines
29 KiB
Nim
Raw Blame History

import
std/[deques, tables, hashes, options, strformat],
chronos, web3, web3/ethtypes as web3Types, json, chronicles,
eth/common/eth_types, eth/async_utils,
spec/[datatypes, digest, crypto, beaconstate, helpers, validator],
network_metadata, merkle_minimal
from times import epochTime
export
web3Types
contract(DepositContract):
proc deposit(pubkey: Bytes48,
withdrawalCredentials: Bytes32,
signature: Bytes96,
deposit_data_root: FixedBytes[32])
proc get_deposit_root(): FixedBytes[32]
proc get_deposit_count(): Bytes8
proc DepositEvent(pubkey: Bytes48,
withdrawalCredentials: Bytes32,
amount: Bytes8,
signature: Bytes96,
index: Bytes8) {.event.}
# TODO
# The raises list of this module are still not usable due to general
# Exceptions being reported from Chronos's asyncfutures2.
type
Eth1BlockNumber* = uint64
Eth1BlockTimestamp* = uint64
Eth1BlockHeader = web3Types.BlockHeader
Eth1Block* = ref object
number*: Eth1BlockNumber
timestamp*: Eth1BlockTimestamp
deposits*: seq[Deposit]
voteData*: Eth1Data
knownGoodDepositsCount*: Option[uint64]
Eth1Chain* = object
knownStart: Eth1Data
knownStartBlockNum: Option[Eth1BlockNumber]
blocks: Deque[Eth1Block]
blocksByHash: Table[BlockHash, Eth1Block]
allDeposits*: seq[Deposit]
MainchainMonitor* = ref object
preset: RuntimePreset
depositContractAddress: Address
dataProviderFactory*: DataProviderFactory
genesisState: NilableBeaconStateRef
genesisStateFut: Future[void]
genesisMonitoringFut: Future[void]
eth1Chain: Eth1Chain
depositQueue: AsyncQueue[Eth1BlockHeader]
runFut: Future[void]
DataProvider* = object of RootObj
DataProviderRef* = ref DataProvider
DataProviderFactory* = object
desc: string
new: proc(depositContractAddress: Address): Future[DataProviderRef] {.
gcsafe
# raises: [Defect]
.}
Web3DataProvider* = object of DataProvider
url: string
web3: Web3
ns: Sender[DepositContract]
blockHeadersSubscription: Subscription
Web3DataProviderRef* = ref Web3DataProvider
ReorgDepthLimitExceeded = object of CatchableError
CorruptDataProvider = object of CatchableError
DisconnectHandler* = proc () {.gcsafe, raises: [Defect].}
DepositEventHandler* = proc (
pubkey: Bytes48,
withdrawalCredentials: Bytes32,
amount: Bytes8,
signature: Bytes96, merkleTreeIndex: Bytes8, j: JsonNode) {.raises: [Defect], gcsafe.}
const
web3Timeouts = 5.seconds
# TODO: Add preset validation
# MIN_GENESIS_ACTIVE_VALIDATOR_COUNT should be larger than SLOTS_PER_EPOCH
# doAssert SECONDS_PER_ETH1_BLOCK * preset.ETH1_FOLLOW_DISTANCE < GENESIS_DELAY,
# "Invalid configuration: GENESIS_DELAY is set too low"
# TODO Nim's analysis on the lock level of the methods in this
# module seems broken. Investigate and file this as an issue.
{.push warning[LockLevel]: off.}
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.2/specs/phase0/validator.md#get_eth1_data
func compute_time_at_slot(state: BeaconState, slot: Slot): uint64 =
state.genesis_time + slot * SECONDS_PER_SLOT
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.2/specs/phase0/validator.md#get_eth1_data
func voting_period_start_time*(state: BeaconState): uint64 =
let eth1_voting_period_start_slot =
state.slot - state.slot mod SLOTS_PER_ETH1_VOTING_PERIOD.uint64
compute_time_at_slot(state, eth1_voting_period_start_slot)
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.2/specs/phase0/validator.md#get_eth1_data
func is_candidate_block(preset: RuntimePreset, blk: Eth1Block, period_start: uint64): bool =
(blk.timestamp + SECONDS_PER_ETH1_BLOCK.uint64 * preset.ETH1_FOLLOW_DISTANCE <= period_start) and
(blk.timestamp + SECONDS_PER_ETH1_BLOCK.uint64 * preset.ETH1_FOLLOW_DISTANCE * 2 >= period_start)
func asEth2Digest*(x: BlockHash): Eth2Digest =
Eth2Digest(data: array[32, byte](x))
template asBlockHash(x: Eth2Digest): BlockHash =
BlockHash(x.data)
func shortLog*(b: Eth1Block): string =
&"{b.number}:{shortLog b.voteData.block_hash}"
func getDepositsInRange(eth1Chain: Eth1Chain,
sinceBlock, latestBlock: Eth1BlockNumber): seq[Deposit] =
## Returns all deposits that happened AFTER the block `sinceBlock` (not inclusive).
## The deposits in `latestBlock` will be included.
if latestBlock <= sinceBlock: return
let firstBlockInCache = eth1Chain.blocks[0].number
# This function should be used with indices obtained with `eth1Chain.findBlock`.
# This guarantess that both of these indices will be valid:
doAssert sinceBlock >= firstBlockInCache and
(latestBlock - firstBlockInCache) < eth1Chain.blocks.lenu64
let
sinceBlockIdx = sinceBlock - firstBlockInCache
latestBlockIdx = latestBlock - firstBlockInCache
for i in (sinceBlockIdx + 1) ..< latestBlockIdx:
result.add eth1Chain.blocks[i].deposits
template findBlock*(eth1Chain: Eth1Chain, hash: BlockHash): Eth1Block =
eth1Chain.blocksByHash.getOrDefault(hash, nil)
template findBlock*(eth1Chain: Eth1Chain, eth1Data: Eth1Data): Eth1Block =
getOrDefault(eth1Chain.blocksByHash, asBlockHash(eth1Data.block_hash), nil)
proc findParent*(eth1Chain: Eth1Chain, blk: BlockObject): Eth1Block =
result = eth1Chain.findBlock(blk.parentHash)
# a distinct type is stipped here:
let blockNumber = Eth1BlockNumber(blk.number)
if result != nil and result.number != blockNumber - 1:
debug "Found inconsistent numbering of Eth1 blocks. Ignoring block.",
blockHash = blk.hash.toHex, blockNumber,
parentHash = blk.parentHash.toHex, parentNumber = result.number
result = nil
func latestCandidateBlock(eth1Chain: Eth1Chain, preset: RuntimePreset, periodStart: uint64): Eth1Block =
for i in countdown(eth1Chain.blocks.len - 1, 0):
let blk = eth1Chain.blocks[i]
if is_candidate_block(preset, blk, periodStart):
return blk
func popBlock(eth1Chain: var Eth1Chain) =
let removed = eth1Chain.blocks.popLast
eth1Chain.blocksByHash.del removed.voteData.block_hash.asBlockHash
func trimHeight(eth1Chain: var Eth1Chain, blockNumber: Eth1BlockNumber) =
## Removes all blocks above certain `blockNumber`
while eth1Chain.blocks.len > 0:
if eth1Chain.blocks.peekLast.number > blockNumber:
eth1Chain.popBlock()
else:
break
if eth1Chain.blocks.len > 0:
eth1Chain.allDeposits.setLen(eth1Chain.blocks[^1].voteData.deposit_count)
else:
eth1Chain.allDeposits.setLen(0)
func isSuccessorBlock(eth1Chain: Eth1Chain, newBlock: Eth1Block): bool =
let currentDepositCount = if eth1Chain.blocks.len == 0:
eth1Chain.knownStart.deposit_count
else:
let lastBlock = eth1Chain.blocks.peekLast
if lastBlock.number >= newBlock.number: return false
lastBlock.voteData.deposit_count
(currentDepositCount + newBlock.deposits.lenu64) == newBlock.voteData.deposit_count
func addSuccessorBlock*(eth1Chain: var Eth1Chain, newBlock: Eth1Block): bool =
result = isSuccessorBlock(eth1Chain, newBlock)
if result:
eth1Chain.allDeposits.add newBlock.deposits
reset newBlock.deposits
eth1Chain.blocks.addLast newBlock
eth1Chain.blocksByHash[newBlock.voteData.block_hash.asBlockHash] = newBlock
func totalDeposits*(eth1Chain: Eth1Chain): int =
for blk in eth1Chain.blocks:
result += blk.deposits.len
func allDeposits*(eth1Chain: Eth1Chain): seq[Deposit] =
for blk in eth1Chain.blocks:
result.add blk.deposits
func clear*(eth1Chain: var Eth1Chain) =
eth1Chain = default(Eth1Chain)
template hash*(x: Eth1Block): Hash =
hash(x.voteData.block_hash.data)
template notImplemented =
doAssert false, "Method not implemented"
method getBlockByHash*(p: DataProviderRef, hash: BlockHash): Future[BlockObject] {.
base
gcsafe
locks: 0
# raises: [Defect]
.} =
notImplemented
method getBlockByNumber*(p: DataProviderRef, hash: Eth1BlockNumber): Future[BlockObject] {.
base
gcsafe
locks: 0
# raises: [Defect]
.} =
notImplemented
method onDisconnect*(p: DataProviderRef, handler: DisconnectHandler) {.
base
gcsafe
locks: 0
# raises: []
.} =
notImplemented
method onBlockHeaders*(p: DataProviderRef,
blockHeaderHandler: BlockHeaderHandler,
errorHandler: SubscriptionErrorHandler): Future[void] {.
base
gcsafe
locks: 0
# raises: []
.} =
notImplemented
method close*(p: DataProviderRef): Future[void] {.
base
gcsafe
locks: 0
# raises: [Defect]
.} =
notImplemented
method fetchDepositData*(p: DataProviderRef,
fromBlock, toBlock: Eth1BlockNumber): Future[seq[Eth1Block]] {.
base
gcsafe
locks: 0
# raises: [Defect, CatchableError]
.} =
notImplemented
method fetchBlockDetails(p: DataProviderRef, blk: Eth1Block): Future[void] {.
base
gcsafe
locks: 0
# raises: [Defect, CatchableError]
.} =
notImplemented
proc new*(T: type Web3DataProvider,
web3Url: string,
depositContractAddress: Address): Future[ref Web3DataProvider] {.
async
# raises: [Defect]
.} =
try:
type R = ref T
let
web3 = await newWeb3(web3Url)
ns = web3.contractSender(DepositContract, depositContractAddress)
return R(url: web3Url, web3: web3, ns: ns)
except CatchableError:
return nil
func web3Provider*(web3Url: string): DataProviderFactory =
proc factory(depositContractAddress: Address): Future[DataProviderRef] {.async.} =
result = await Web3DataProvider.new(web3Url, depositContractAddress)
DataProviderFactory(desc: "web3(" & web3Url & ")", new: factory)
method close*(p: Web3DataProviderRef): Future[void] {.async, locks: 0.} =
if p.blockHeadersSubscription != nil:
await p.blockHeadersSubscription.unsubscribe()
await p.web3.close()
method getBlockByHash*(p: Web3DataProviderRef, hash: BlockHash): Future[BlockObject] =
return p.web3.provider.eth_getBlockByHash(hash, false)
method getBlockByNumber*(p: Web3DataProviderRef, number: Eth1BlockNumber): Future[BlockObject] =
return p.web3.provider.eth_getBlockByNumber(&"0x{number:X}", false)
proc getBlockNumber(p: DataProviderRef, hash: BlockHash): Future[Eth1BlockNumber] {.async.} =
try:
let blk = awaitWithTimeout(p.getBlockByHash(hash), web3Timeouts):
return 0
return Eth1BlockNumber(blk.number)
except CatchableError as exc:
notice "Failed to get Eth1 block number from hash",
hash = $hash, err = exc.msg
raise exc
template readJsonField(j: JsonNode,
fieldName: string,
ValueType: type): untyped =
var res: ValueType
fromJson(j[fieldName], fieldName, res)
res
proc readJsonDeposits(depositsList: JsonNode): seq[Eth1Block] =
if depositsList.kind != JArray:
raise newException(CatchableError,
"Web3 provider didn't return a list of deposit events")
var lastEth1Block: Eth1Block
for logEvent in depositsList:
let
blockNumber = Eth1BlockNumber readJsonField(logEvent, "blockNumber", Quantity)
blockHash = readJsonField(logEvent, "blockHash", BlockHash)
logData = strip0xPrefix(logEvent["data"].getStr)
if lastEth1Block == nil or lastEth1Block.number != blockNumber:
lastEth1Block = Eth1Block(
number: blockNumber,
voteData: Eth1Data(block_hash: blockHash.asEth2Digest))
result.add lastEth1Block
var
pubkey: Bytes48
withdrawalCredentials: Bytes32
amount: Bytes8
signature: Bytes96
index: Bytes8
var offset = 0
offset += decode(logData, offset, pubkey)
offset += decode(logData, offset, withdrawalCredentials)
offset += decode(logData, offset, amount)
offset += decode(logData, offset, signature)
offset += decode(logData, offset, index)
lastEth1Block.deposits.add Deposit(
data: DepositData(
pubkey: ValidatorPubKey.init(array[48, byte](pubkey)),
withdrawal_credentials: Eth2Digest(data: array[32, byte](withdrawalCredentials)),
amount: bytes_to_uint64(array[8, byte](amount)),
signature: ValidatorSig.init(array[96, byte](signature))))
method fetchDepositData*(p: Web3DataProviderRef,
fromBlock, toBlock: Eth1BlockNumber): Future[seq[Eth1Block]]
{.async, locks: 0.} =
info "Obtaining deposit log events", fromBlock, toBlock
return readJsonDeposits(await p.ns.getJsonLogs(DepositEvent,
fromBlock = some blockId(fromBlock),
toBlock = some blockId(toBlock)))
method fetchBlockDetails(p: Web3DataProviderRef, blk: Eth1Block) {.async.} =
let
web3Block = p.getBlockByNumber(blk.number)
depositRoot = p.ns.get_deposit_root.call(blockNumber = blk.number)
rawCount = p.ns.get_deposit_count.call(blockNumber = blk.number)
discard await web3Block
discard await depositRoot
discard await rawCount
let depositCount = bytes_to_uint64(array[8, byte](rawCount.read))
blk.timestamp = Eth1BlockTimestamp(web3Block.read.timestamp)
blk.voteData.deposit_count = depositCount
blk.voteData.deposit_root = depositRoot.read.asEth2Digest
method onDisconnect*(p: Web3DataProviderRef, handler: DisconnectHandler) {.
gcsafe
locks: 0
# raises: []
.} =
p.web3.onDisconnect = handler
method onBlockHeaders*(p: Web3DataProviderRef,
blockHeaderHandler: BlockHeaderHandler,
errorHandler: SubscriptionErrorHandler): Future[void] {.
async
gcsafe
locks: 0
# raises: []
.} =
if p.blockHeadersSubscription != nil:
await p.blockHeadersSubscription.unsubscribe()
info "Waiting for new Eth1 block headers"
let options = newJObject()
p.blockHeadersSubscription = await p.web3.subscribeForBlockHeaders(
options, blockHeaderHandler, errorHandler)
# https://github.com/ethereum/eth2.0-specs/blob/v0.11.1/specs/phase0/validator.md#get_eth1_data
func getBlockProposalData*(preset: RuntimePreset, eth1Chain: Eth1Chain,
state: BeaconState): (Eth1Data, seq[Deposit]) =
template voteForNoChange() =
return (state.eth1_data, newSeq[Deposit]())
let prevBlock = eth1Chain.findBlock(state.eth1_data)
if prevBlock == nil:
# The Eth1 block currently referenced in the BeaconState is unknown to us.
# This situation is not specifically covered in the honest validator spec,
# but there is a similar condition where none of the eth1_data_votes is
# present in our worldview. The suggestion there is to vote for "no change"
# and we'll do the same here:
voteForNoChange()
let periodStart = voting_period_start_time(state)
var otherVotesCountTable = initCountTable[Eth1Block]()
for vote in state.eth1_data_votes:
let eth1Block = eth1Chain.findBlock(vote)
if eth1Block != nil and is_candidate_block(preset, eth1Block, periodStart):
otherVotesCountTable.inc eth1Block
var ourVote: Eth1Block
if otherVotesCountTable.len > 0:
ourVote = otherVotesCountTable.largest.key
else:
ourVote = eth1Chain.latestCandidateBlock(preset, periodStart)
if ourVote == nil:
voteForNoChange()
(ourVote.voteData, eth1Chain.getDepositsInRange(prevBlock.number, ourVote.number))
template getBlockProposalData*(m: MainchainMonitor, state: BeaconState): untyped =
getBlockProposalData(m.preset, m.eth1Chain, state)
proc init*(T: type MainchainMonitor,
preset: RuntimePreset,
dataProviderFactory: DataProviderFactory,
depositContractAddress: Eth1Address,
startPosition: Eth1Data): T =
T(preset: preset,
depositQueue: newAsyncQueue[Eth1BlockHeader](),
dataProviderFactory: dataProviderFactory,
depositContractAddress: Address depositContractAddress,
eth1Chain: Eth1Chain(knownStart: startPosition))
proc isCandidateForGenesis(preset: RuntimePreset,
timeNow: float,
blk: Eth1Block): bool =
let followDistanceInSeconds = uint64(SECONDS_PER_ETH1_BLOCK) * preset.ETH1_FOLLOW_DISTANCE
if float(blk.timestamp + followDistanceInSeconds) > timeNow:
return false
if genesis_time_from_eth1_timestamp(preset, blk.timestamp) < preset.MIN_GENESIS_TIME:
return false
if blk.knownGoodDepositsCount.isSome:
blk.knownGoodDepositsCount.get >= preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT
else:
blk.voteData.deposit_count >= preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT
proc minGenesisCandidateBlockIdx(m: MainchainMonitor): Option[int]
{.raises: [Defect].} =
if m.eth1Chain.blocks.len == 0:
return
let now = epochTime()
if not isCandidateForGenesis(m.preset, now, m.eth1Chain.blocks.peekLast):
return
var candidatePos = m.eth1Chain.blocks.len - 1
while candidatePos > 1:
if not isCandidateForGenesis(m.preset, now, m.eth1Chain.blocks[candidatePos - 1]):
break
dec candidatePos
return some(candidatePos)
proc createBeaconStateAux(preset: RuntimePreset,
eth1Block: Eth1Block,
deposits: var openarray[Deposit]): BeaconStateRef =
attachMerkleProofs deposits
result = initialize_beacon_state_from_eth1(preset,
eth1Block.voteData.block_hash,
eth1Block.timestamp.uint64,
deposits, {})
var cache = StateCache()
let activeValidators = count_active_validators(result[], GENESIS_EPOCH, cache)
eth1Block.knownGoodDepositsCount = some activeValidators
proc createBeaconState(m: MainchainMonitor, eth1Block: Eth1Block): BeaconStateRef =
createBeaconStateAux(
m.preset,
eth1Block,
m.eth1Chain.allDeposits.toOpenArray(0, int(eth1Block.voteData.deposit_count - 1)))
proc signalGenesis(m: MainchainMonitor, genesisState: BeaconStateRef) =
m.genesisState = genesisState
if not m.genesisStateFut.isNil:
m.genesisStateFut.complete()
m.genesisStateFut = nil
proc findGenesisBlockInRange(m: MainchainMonitor,
startBlock, endBlock: Eth1Block): Future[Eth1Block]
{.async.} =
let dataProvider = await m.dataProviderFactory.new(m.depositContractAddress)
if dataProvider == nil:
error "Failed to initialize Eth1 data provider",
provider = m.dataProviderFactory.desc
raise newException(CatchableError, "Failed to initialize Eth1 data provider")
var
startBlock = startBlock
endBlock = endBlock
depositData = startBlock.voteData
while startBlock.number + 1 < endBlock.number:
let
MIN_GENESIS_TIME = m.preset.MIN_GENESIS_TIME
startBlockTime = genesis_time_from_eth1_timestamp(m.preset, startBlock.timestamp)
secondsPerBlock = float(endBlock.timestamp - startBlock.timestamp) /
float(endBlock.number - startBlock.number)
blocksToJump = max(float(MIN_GENESIS_TIME - startBlockTime) / secondsPerBlock, 1.0)
candidateNumber = min(endBlock.number - 1, startBlock.number + 1) # blocksToJump.uint64)
candidateBlock = await dataProvider.getBlockByNumber(candidateNumber)
var candidateAsEth1Block = Eth1Block(number: candidateBlock.number.uint64,
timestamp: candidateBlock.timestamp.uint64,
voteData: depositData)
candidateAsEth1Block.voteData.block_hash = candidateBlock.hash.asEth2Digest
let candidateGenesisTime = genesis_time_from_eth1_timestamp(
m.preset, candidateBlock.timestamp.uint64)
info "Probing possible genesis block",
`block` = candidateBlock.number.uint64,
candidateGenesisTime
if candidateGenesisTime < MIN_GENESIS_TIME:
startBlock = candidateAsEth1Block
else:
endBlock = candidateAsEth1Block
return endBlock
proc checkForGenesisLoop(m: MainchainMonitor) {.async.} =
while true:
if not m.genesisState.isNil:
return
try:
let genesisCandidateIdx = m.minGenesisCandidateBlockIdx
if genesisCandidateIdx.isSome:
let
genesisCandidateIdx = genesisCandidateIdx.get
genesisCandidate = m.eth1Chain.blocks[genesisCandidateIdx]
info "Generating state for candidate block for genesis",
blockNum = genesisCandidate.number,
blockHash = genesisCandidate.voteData.block_hash,
potentialDeposits = genesisCandidate.voteData.deposit_count
let
candidateState = m.createBeaconState(genesisCandidate)
if genesisCandidate.knownGoodDepositsCount.get >= m.preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT:
# We have a candidate state on our hands, but our current Eth1Chain
# may consist only of blocks that have deposits attached to them
# while the real genesis may have happened in a block without any
# deposits (triggered by MIN_GENESIS_TIME).
#
# This can happen when the beacon node is launched after the genesis
# event. We take a short cut when constructing the initial Eth1Chain
# by downloading only deposit log entries. Thus, we'll see all the
# blocks with deposits, but not the regular blocks in between.
#
# We'll handle this special case below by examing whether we are in
# this potential scenario and we'll use a fast guessing algorith to
# discover the ETh1 block with minimal valid genesis time.
if genesisCandidateIdx > 0:
let preceedingEth1Block = m.eth1Chain.blocks[genesisCandidateIdx - 1]
if preceedingEth1Block.voteData.deposit_root == genesisCandidate.voteData.deposit_root:
preceedingEth1Block.knownGoodDepositsCount = genesisCandidate.knownGoodDepositsCount
else:
discard m.createBeaconState(preceedingEth1Block)
if preceedingEth1Block.knownGoodDepositsCount.get >= m.preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT and
genesisCandidate.number - preceedingEth1Block.number > 1:
let genesisBlock = await m.findGenesisBlockInRange(preceedingEth1Block, genesisCandidate)
if genesisBlock.number != genesisCandidate.number:
m.signalGenesis m.createBeaconState(genesisBlock)
return
m.signalGenesis candidateState
return
else:
info "Eth2 genesis candidate block rejected",
`block` = shortLog(genesisCandidate),
validDeposits = genesisCandidate.knownGoodDepositsCount.get,
needed = m.preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT
else:
# TODO: check for a stale monitor
discard
except CatchableError as err:
debug "Unexpected error in checkForGenesisLoop", err = err.msg
await sleepAsync(1.seconds)
proc waitGenesis*(m: MainchainMonitor): Future[BeaconStateRef] {.async.} =
if m.genesisState.isNil:
if m.genesisStateFut.isNil:
m.genesisStateFut = newFuture[void]("waitGenesis")
m.genesisMonitoringFut = m.checkForGenesisLoop()
await m.genesisStateFut
m.genesisStateFut = nil
if m.genesisState != nil:
return m.genesisState
else:
result = new BeaconStateRef # make the compiler happy
raiseAssert "Unreachable code"
func totalNonFinalizedBlocks(eth1Chain: Eth1Chain): Natural =
# TODO: implement this precisely
eth1Chain.blocks.len
func latestEth1Data(eth1Chain: Eth1Chain): Eth1Data =
if eth1Chain.blocks.len > 0:
eth1Chain.blocks[^1].voteData
else:
eth1Chain.knownStart
func knownInvalidDepositsCount(eth1Chain: Eth1Chain): uint64 =
for i in countdown(eth1Chain.blocks.len - 1, 0):
let blk = eth1Chain.blocks[i]
if blk.knownGoodDepositsCount.isSome:
return blk.voteData.deposit_count - blk.knownGoodDepositsCount.get
return 0
func maxValidDeposits(eth1Chain: Eth1Chain): uint64 =
if eth1Chain.blocks.len > 0:
let lastBlock = eth1Chain.blocks[^1]
lastBlock.knownGoodDepositsCount.get(
lastBlock.voteData.deposit_count - eth1Chain.knownInvalidDepositsCount)
else:
0
proc processDeposits(m: MainchainMonitor,
dataProvider: DataProviderRef) {.async.} =
# ATTENTION!
# Please note that this code is using a queue to guarantee the
# strict serial order of processing of deposits. If we had the
# same code embedded in the deposit contracts events handler,
# it could easily re-order the steps due to the intruptable
# interleaved execution of async code.
while true:
let blk = await m.depositQueue.popFirst()
m.eth1Chain.trimHeight(Eth1BlockNumber(blk.number) - 1)
let latestKnownBlock = if m.eth1Chain.blocks.len > 0:
m.eth1Chain.blocks[^1].number
elif m.eth1Chain.knownStartBlockNum.isSome:
m.eth1Chain.knownStartBlockNum.get
else:
m.eth1Chain.knownStartBlockNum = some(
await dataProvider.getBlockNumber(m.eth1Chain.knownStart.block_hash.asBlockHash))
m.eth1Chain.knownStartBlockNum.get
let eth1Blocks = await dataProvider.fetchDepositData(latestKnownBlock + 1,
Eth1BlockNumber blk.number)
if eth1Blocks.len == 0:
if m.eth1Chain.maxValidDeposits > m.preset.MIN_GENESIS_ACTIVE_VALIDATOR_COUNT and
m.eth1Chain.knownStart.deposit_count == 0:
let latestEth1Data = m.eth1Chain.latestEth1Data
for missingBlockNum in latestKnownBlock + 1 ..< Eth1BlockNumber(blk.number):
let missingBlock = await dataProvider.getBlockByNumber(missingBlockNum)
doAssert m.eth1Chain.addSuccessorBlock Eth1Block(
number: Eth1BlockNumber(missingBlock.number),
timestamp: Eth1BlockTimestamp(missingBlock.timestamp),
voteData: latestEth1Data)
doAssert m.eth1Chain.addSuccessorBlock Eth1Block(
number: Eth1BlockNumber(blk.number),
timestamp: Eth1BlockTimestamp(blk.timestamp),
voteData: latestEth1Data)
else:
template logBlockProcessed(blk) =
info "Eth1 block processed",
`block` = shortLog(blk), totalDeposits = blk.voteData.deposit_count
await dataProvider.fetchBlockDetails(eth1Blocks[0])
if m.eth1Chain.addSuccessorBlock(eth1Blocks[0]):
logBlockProcessed eth1Blocks[0]
for i in 1 ..< eth1Blocks.len:
await dataProvider.fetchBlockDetails(eth1Blocks[i])
if m.eth1Chain.addSuccessorBlock(eth1Blocks[i]):
logBlockProcessed eth1Blocks[i]
else:
raise newException(CorruptDataProvider,
"A non-successor Eth1 block reported")
else:
# A non-continuous chain detected.
# This could be the result of a deeper fork that was not reported
# properly by the web3 provider. Since this should be an extremely
# rare event we can afford to handle it in a relatively inefficient
# manner. Let's delete half of our non-finalized chain and try again.
var blocksToPop = 0
if m.eth1Chain.blocks.len > 0:
blocksToPop = max(1, m.eth1Chain.totalNonFinalizedBlocks div 2)
for i in 0 ..< blocksToPop:
m.eth1Chain.popBlock()
warn "Web3 provider responded with a non-continous chain of deposits",
backtrackedDeposits = blocksToPop
m.depositQueue.addFirstNoWait blk
proc isRunning*(m: MainchainMonitor): bool =
not m.runFut.isNil
func `===`(json: JsonNode, boolean: bool): bool =
json.kind == JBool and json.bval == boolean
proc run(m: MainchainMonitor, delayBeforeStart: Duration) {.async.} =
if delayBeforeStart != ZeroDuration:
await sleepAsync(delayBeforeStart)
let dataProvider = await m.dataProviderFactory.new(m.depositContractAddress)
if dataProvider == nil:
error "Failed to initialize Eth1 data provider",
provider = m.dataProviderFactory.desc
raise newException(CatchableError, "Failed to initialize Eth1 data provider")
try:
info "Starting Eth1 deposit contract monitoring",
contract = $m.depositContractAddress,
url = m.dataProviderFactory.desc
await dataProvider.onBlockHeaders do (blk: Eth1BlockHeader)
{.raises: [Defect], gcsafe.}:
try:
m.depositQueue.addLastNoWait(blk)
except AsyncQueueFullError:
raiseAssert "The depositQueue has no size limit"
except Exception:
# TODO Investigate why this exception is being raised
raiseAssert "queue.addLastNoWait should not raise exceptions"
do (err: CatchableError):
debug "Error while processing Eth1 block headers subscription", err = err.msg
await m.processDeposits(dataProvider)
finally:
await close(dataProvider)
proc safeCancel(fut: var Future[void]) =
if not fut.isNil and not fut.finished:
fut.cancel()
fut = nil
proc stop*(m: MainchainMonitor) =
safeCancel m.runFut
safeCancel m.genesisMonitoringFut
proc start(m: MainchainMonitor, delayBeforeStart: Duration) =
if m.runFut.isNil:
let runFut = m.run(delayBeforeStart)
m.runFut = runFut
runFut.addCallback do (p: pointer):
if runFut.failed:
if runFut.error[] of CatchableError:
if runFut == m.runFut:
error "Mainchain monitor failure, restarting", err = runFut.error.msg
m.stop()
m.start(5.seconds)
else:
fatal "Fatal exception reached", err = runFut.error.msg
quit 1
proc start*(m: MainchainMonitor) {.inline.} =
m.start(0.seconds)
proc getEth1BlockHash*(url: string, blockId: RtBlockIdentifier): Future[BlockHash] {.async.} =
let web3 = await newWeb3(url)
try:
let blk = await web3.provider.eth_getBlockByNumber(blockId, false)
return blk.hash
finally:
await web3.close()
{.pop.}
Reference in New Issue View Git Blame Copy Permalink