You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
379 lines
13 KiB
379 lines
13 KiB
package node
|
|
|
|
import (
|
|
"errors"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/harmony-one/harmony/consensus"
|
|
|
|
"github.com/harmony-one/harmony/crypto/bls"
|
|
|
|
staking "github.com/harmony-one/harmony/staking/types"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/harmony-one/harmony/core/rawdb"
|
|
"github.com/harmony-one/harmony/core/types"
|
|
"github.com/harmony-one/harmony/internal/utils"
|
|
"github.com/harmony-one/harmony/shard"
|
|
)
|
|
|
|
// Constants of proposing a new block
|
|
const (
|
|
SleepPeriod = 20 * time.Millisecond
|
|
IncomingReceiptsLimit = 6000 // 2000 * (numShards - 1)
|
|
)
|
|
|
|
// WaitForConsensusReadyV2 listen for the readiness signal from consensus and generate new block for consensus.
|
|
// only leader will receive the ready signal
|
|
func (node *Node) WaitForConsensusReadyV2(readySignal chan consensus.ProposalType, commitSigsChan chan []byte, stopChan chan struct{}, stoppedChan chan struct{}) {
|
|
go func() {
|
|
// Setup stoppedChan
|
|
defer close(stoppedChan)
|
|
|
|
utils.Logger().Debug().
|
|
Msg("Waiting for Consensus ready")
|
|
select {
|
|
case <-time.After(30 * time.Second):
|
|
case <-stopChan:
|
|
return
|
|
}
|
|
|
|
for {
|
|
// keep waiting for Consensus ready
|
|
select {
|
|
case <-stopChan:
|
|
utils.Logger().Warn().
|
|
Msg("Consensus new block proposal: STOPPED!")
|
|
return
|
|
case proposalType := <-readySignal:
|
|
retryCount := 0
|
|
for node.Consensus != nil && node.Consensus.IsLeader() {
|
|
time.Sleep(SleepPeriod)
|
|
utils.Logger().Info().
|
|
Uint64("blockNum", node.Blockchain().CurrentBlock().NumberU64()+1).
|
|
Bool("asyncProposal", proposalType == consensus.AsyncProposal).
|
|
Msg("PROPOSING NEW BLOCK ------------------------------------------------")
|
|
|
|
// Prepare last commit signatures
|
|
newCommitSigsChan := make(chan []byte)
|
|
|
|
go func() {
|
|
waitTime := 0 * time.Second
|
|
if proposalType == consensus.AsyncProposal {
|
|
waitTime = consensus.CommitSigReceiverTimeout
|
|
}
|
|
select {
|
|
case <-time.After(waitTime):
|
|
if waitTime == 0 {
|
|
utils.Logger().Info().Msg("[ProposeNewBlock] Sync block proposal, reading commit sigs directly from DB")
|
|
} else {
|
|
utils.Logger().Info().Msg("[ProposeNewBlock] Timeout waiting for commit sigs, reading directly from DB")
|
|
}
|
|
sigs, err := node.Consensus.BlockCommitSigs(node.Blockchain().CurrentBlock().NumberU64())
|
|
|
|
if err != nil {
|
|
utils.Logger().Error().Err(err).Msg("[ProposeNewBlock] Cannot get commit signatures from last block")
|
|
} else {
|
|
newCommitSigsChan <- sigs
|
|
}
|
|
case commitSigs := <-commitSigsChan:
|
|
utils.Logger().Info().Msg("[ProposeNewBlock] received commit sigs asynchronously")
|
|
if len(commitSigs) > bls.BLSSignatureSizeInBytes {
|
|
newCommitSigsChan <- commitSigs
|
|
}
|
|
}
|
|
}()
|
|
newBlock, err := node.ProposeNewBlock(newCommitSigsChan)
|
|
|
|
if err == nil {
|
|
if blk, ok := node.proposedBlock[newBlock.NumberU64()]; ok {
|
|
utils.Logger().Info().Uint64("blockNum", newBlock.NumberU64()).Str("blockHash", blk.Hash().Hex()).
|
|
Msg("Block with the same number was already proposed, abort.")
|
|
break
|
|
}
|
|
utils.Logger().Info().
|
|
Uint64("blockNum", newBlock.NumberU64()).
|
|
Uint64("epoch", newBlock.Epoch().Uint64()).
|
|
Uint64("viewID", newBlock.Header().ViewID().Uint64()).
|
|
Int("numTxs", newBlock.Transactions().Len()).
|
|
Int("numStakingTxs", newBlock.StakingTransactions().Len()).
|
|
Int("crossShardReceipts", newBlock.IncomingReceipts().Len()).
|
|
Msg("=========Successfully Proposed New Block==========")
|
|
|
|
// Send the new block to Consensus so it can be confirmed.
|
|
node.proposedBlock[newBlock.NumberU64()] = newBlock
|
|
delete(node.proposedBlock, newBlock.NumberU64()-10)
|
|
node.BlockChannel <- newBlock
|
|
break
|
|
} else {
|
|
retryCount++
|
|
utils.Logger().Err(err).Int("retryCount", retryCount).
|
|
Msg("!!!!!!!!!Failed Proposing New Block!!!!!!!!!")
|
|
if retryCount > 3 {
|
|
// break to avoid repeated failures
|
|
break
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// ProposeNewBlock proposes a new block...
|
|
func (node *Node) ProposeNewBlock(commitSigs chan []byte) (*types.Block, error) {
|
|
currentHeader := node.Blockchain().CurrentHeader()
|
|
nowEpoch, blockNow := currentHeader.Epoch(), currentHeader.Number()
|
|
utils.AnalysisStart("ProposeNewBlock", nowEpoch, blockNow)
|
|
defer utils.AnalysisEnd("ProposeNewBlock", nowEpoch, blockNow)
|
|
|
|
node.Worker.UpdateCurrent()
|
|
|
|
header := node.Worker.GetCurrentHeader()
|
|
// Update worker's current header and
|
|
// state data in preparation to propose/process new transactions
|
|
leaderKey := node.Consensus.LeaderPubKey
|
|
var (
|
|
coinbase = node.GetAddressForBLSKey(leaderKey.Object, header.Epoch())
|
|
beneficiary = coinbase
|
|
err error
|
|
)
|
|
|
|
// After staking, all coinbase will be the address of bls pub key
|
|
if node.Blockchain().Config().IsStaking(header.Epoch()) {
|
|
blsPubKeyBytes := leaderKey.Object.GetAddress()
|
|
coinbase.SetBytes(blsPubKeyBytes[:])
|
|
}
|
|
|
|
emptyAddr := common.Address{}
|
|
if coinbase == emptyAddr {
|
|
return nil, errors.New("[ProposeNewBlock] Failed setting coinbase")
|
|
}
|
|
|
|
// Must set coinbase here because the operations below depend on it
|
|
header.SetCoinbase(coinbase)
|
|
|
|
// Get beneficiary based on coinbase
|
|
// Before staking, coinbase itself is the beneficial
|
|
// After staking, beneficial is the corresponding ECDSA address of the bls key
|
|
beneficiary, err = node.Blockchain().GetECDSAFromCoinbase(header)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Prepare normal and staking transactions retrieved from transaction pool
|
|
utils.AnalysisStart("proposeNewBlockChooseFromTxnPool")
|
|
|
|
pendingPoolTxs, err := node.TxPool.Pending()
|
|
if err != nil {
|
|
utils.Logger().Err(err).Msg("Failed to fetch pending transactions")
|
|
return nil, err
|
|
}
|
|
pendingPlainTxs := map[common.Address]types.Transactions{}
|
|
pendingStakingTxs := staking.StakingTransactions{}
|
|
for addr, poolTxs := range pendingPoolTxs {
|
|
plainTxsPerAcc := types.Transactions{}
|
|
for _, tx := range poolTxs {
|
|
if plainTx, ok := tx.(*types.Transaction); ok {
|
|
plainTxsPerAcc = append(plainTxsPerAcc, plainTx)
|
|
} else if stakingTx, ok := tx.(*staking.StakingTransaction); ok {
|
|
// Only process staking transactions after pre-staking epoch happened.
|
|
if node.Blockchain().Config().IsPreStaking(node.Worker.GetCurrentHeader().Epoch()) {
|
|
pendingStakingTxs = append(pendingStakingTxs, stakingTx)
|
|
}
|
|
} else {
|
|
utils.Logger().Err(types.ErrUnknownPoolTxType).
|
|
Msg("Failed to parse pending transactions")
|
|
return nil, types.ErrUnknownPoolTxType
|
|
}
|
|
}
|
|
if plainTxsPerAcc.Len() > 0 {
|
|
pendingPlainTxs[addr] = plainTxsPerAcc
|
|
}
|
|
}
|
|
utils.AnalysisEnd("proposeNewBlockChooseFromTxnPool")
|
|
|
|
// Try commit normal and staking transactions based on the current state
|
|
// The successfully committed transactions will be put in the proposed block
|
|
if err := node.Worker.CommitTransactions(
|
|
pendingPlainTxs, pendingStakingTxs, beneficiary,
|
|
); err != nil {
|
|
utils.Logger().Error().Err(err).Msg("cannot commit transactions")
|
|
return nil, err
|
|
}
|
|
|
|
// Prepare cross shard transaction receipts
|
|
receiptsList := node.proposeReceiptsProof()
|
|
if len(receiptsList) != 0 {
|
|
if err := node.Worker.CommitReceipts(receiptsList); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
isBeaconchainInCrossLinkEra := node.NodeConfig.ShardID == shard.BeaconChainShardID &&
|
|
node.Blockchain().Config().IsCrossLink(node.Worker.GetCurrentHeader().Epoch())
|
|
|
|
isBeaconchainInStakingEra := node.NodeConfig.ShardID == shard.BeaconChainShardID &&
|
|
node.Blockchain().Config().IsStaking(node.Worker.GetCurrentHeader().Epoch())
|
|
|
|
utils.AnalysisStart("proposeNewBlockVerifyCrossLinks")
|
|
// Prepare cross links and slashing messages
|
|
var crossLinksToPropose types.CrossLinks
|
|
if isBeaconchainInCrossLinkEra {
|
|
allPending, err := node.Blockchain().ReadPendingCrossLinks()
|
|
invalidToDelete := []types.CrossLink{}
|
|
if err == nil {
|
|
for _, pending := range allPending {
|
|
exist, err := node.Blockchain().ReadCrossLink(pending.ShardID(), pending.BlockNum())
|
|
if err == nil || exist != nil {
|
|
invalidToDelete = append(invalidToDelete, pending)
|
|
utils.Logger().Debug().
|
|
AnErr("[ProposeNewBlock] pending crosslink is already committed onchain", err)
|
|
continue
|
|
}
|
|
|
|
// Crosslink is already verified before it's accepted to pending,
|
|
// no need to verify again in proposal.
|
|
if !node.Blockchain().Config().IsCrossLink(pending.Epoch()) {
|
|
utils.Logger().Debug().
|
|
AnErr("[ProposeNewBlock] pending crosslink that's before crosslink epoch", err)
|
|
continue
|
|
}
|
|
|
|
crossLinksToPropose = append(crossLinksToPropose, pending)
|
|
if len(crossLinksToPropose) > 15 {
|
|
break
|
|
}
|
|
}
|
|
utils.Logger().Info().
|
|
Msgf("[ProposeNewBlock] Proposed %d crosslinks from %d pending crosslinks",
|
|
len(crossLinksToPropose), len(allPending),
|
|
)
|
|
} else {
|
|
utils.Logger().Error().Err(err).Msgf(
|
|
"[ProposeNewBlock] Unable to Read PendingCrossLinks, number of crosslinks: %d",
|
|
len(allPending),
|
|
)
|
|
}
|
|
node.Blockchain().DeleteFromPendingCrossLinks(invalidToDelete)
|
|
}
|
|
utils.AnalysisEnd("proposeNewBlockVerifyCrossLinks")
|
|
|
|
if isBeaconchainInStakingEra {
|
|
// this will set a meaningful w.current.slashes
|
|
if err := node.Worker.CollectVerifiedSlashes(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// Prepare shard state
|
|
var shardState *shard.State
|
|
if shardState, err = node.Blockchain().SuperCommitteeForNextEpoch(
|
|
node.Beaconchain(), node.Worker.GetCurrentHeader(), false,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
viewIDFunc := func() uint64 {
|
|
return node.Consensus.GetCurBlockViewID()
|
|
}
|
|
finalizedBlock, err := node.Worker.FinalizeNewBlock(
|
|
commitSigs, viewIDFunc,
|
|
coinbase, crossLinksToPropose, shardState,
|
|
)
|
|
if err != nil {
|
|
utils.Logger().Error().Err(err).Msg("[ProposeNewBlock] Failed finalizing the new block")
|
|
return nil, err
|
|
}
|
|
utils.Logger().Info().Msg("[ProposeNewBlock] verifying the new block header")
|
|
err = node.Blockchain().Validator().ValidateHeader(finalizedBlock, true)
|
|
|
|
if err != nil {
|
|
utils.Logger().Error().Err(err).Msg("[ProposeNewBlock] Failed verifying the new block header")
|
|
return nil, err
|
|
}
|
|
return finalizedBlock, nil
|
|
}
|
|
|
|
func (node *Node) proposeReceiptsProof() []*types.CXReceiptsProof {
|
|
if !node.Blockchain().Config().HasCrossTxFields(node.Worker.GetCurrentHeader().Epoch()) {
|
|
return []*types.CXReceiptsProof{}
|
|
}
|
|
|
|
numProposed := 0
|
|
validReceiptsList := []*types.CXReceiptsProof{}
|
|
pendingReceiptsList := []*types.CXReceiptsProof{}
|
|
|
|
node.pendingCXMutex.Lock()
|
|
defer node.pendingCXMutex.Unlock()
|
|
|
|
// not necessary to sort the list, but we just prefer to process the list ordered by shard and blocknum
|
|
pendingCXReceipts := []*types.CXReceiptsProof{}
|
|
for _, v := range node.pendingCXReceipts {
|
|
pendingCXReceipts = append(pendingCXReceipts, v)
|
|
}
|
|
|
|
sort.SliceStable(pendingCXReceipts, func(i, j int) bool {
|
|
shardCMP := pendingCXReceipts[i].MerkleProof.ShardID < pendingCXReceipts[j].MerkleProof.ShardID
|
|
shardEQ := pendingCXReceipts[i].MerkleProof.ShardID == pendingCXReceipts[j].MerkleProof.ShardID
|
|
blockCMP := pendingCXReceipts[i].MerkleProof.BlockNum.Cmp(
|
|
pendingCXReceipts[j].MerkleProof.BlockNum,
|
|
) == -1
|
|
return shardCMP || (shardEQ && blockCMP)
|
|
})
|
|
|
|
m := map[common.Hash]struct{}{}
|
|
|
|
Loop:
|
|
for _, cxp := range node.pendingCXReceipts {
|
|
if numProposed > IncomingReceiptsLimit {
|
|
pendingReceiptsList = append(pendingReceiptsList, cxp)
|
|
continue
|
|
}
|
|
// check double spent
|
|
if node.Blockchain().IsSpent(cxp) {
|
|
utils.Logger().Debug().Interface("cxp", cxp).Msg("[proposeReceiptsProof] CXReceipt is spent")
|
|
continue
|
|
}
|
|
hash := cxp.MerkleProof.BlockHash
|
|
// ignore duplicated receipts
|
|
if _, ok := m[hash]; ok {
|
|
continue
|
|
} else {
|
|
m[hash] = struct{}{}
|
|
}
|
|
|
|
for _, item := range cxp.Receipts {
|
|
if item.ToShardID != node.Blockchain().ShardID() {
|
|
continue Loop
|
|
}
|
|
}
|
|
|
|
if err := node.Blockchain().Validator().ValidateCXReceiptsProof(cxp); err != nil {
|
|
if strings.Contains(err.Error(), rawdb.MsgNoShardStateFromDB) {
|
|
pendingReceiptsList = append(pendingReceiptsList, cxp)
|
|
} else {
|
|
utils.Logger().Error().Err(err).Msg("[proposeReceiptsProof] Invalid CXReceiptsProof")
|
|
}
|
|
continue
|
|
}
|
|
|
|
utils.Logger().Debug().Interface("cxp", cxp).Msg("[proposeReceiptsProof] CXReceipts Added")
|
|
validReceiptsList = append(validReceiptsList, cxp)
|
|
numProposed = numProposed + len(cxp.Receipts)
|
|
}
|
|
|
|
node.pendingCXReceipts = make(map[string]*types.CXReceiptsProof)
|
|
for _, v := range pendingReceiptsList {
|
|
blockNum := v.Header.Number().Uint64()
|
|
shardID := v.Header.ShardID()
|
|
key := utils.GetPendingCXKey(shardID, blockNum)
|
|
node.pendingCXReceipts[key] = v
|
|
}
|
|
|
|
utils.Logger().Debug().Msgf("[proposeReceiptsProof] number of validReceipts %d", len(validReceiptsList))
|
|
return validReceiptsList
|
|
}
|
|
|