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.
252 lines
8.5 KiB
252 lines
8.5 KiB
package node
|
|
|
|
import (
|
|
"math/big"
|
|
"sort"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
|
|
"github.com/ethereum/go-ethereum/rlp"
|
|
|
|
"github.com/harmony-one/harmony/core"
|
|
"github.com/harmony-one/harmony/core/types"
|
|
"github.com/harmony-one/harmony/internal/ctxerror"
|
|
"github.com/harmony-one/harmony/internal/utils"
|
|
"github.com/harmony-one/harmony/shard"
|
|
)
|
|
|
|
// Constants of lower bound limit of a new block.
|
|
const (
|
|
PeriodicBlock = 200 * time.Millisecond
|
|
)
|
|
|
|
// WaitForConsensusReadyv2 listen for the readiness signal from consensus and generate new block for consensus.
|
|
// only leader will receive the ready signal
|
|
// TODO: clean pending transactions for validators; or validators not prepare pending transactions
|
|
func (node *Node) WaitForConsensusReadyv2(readySignal chan struct{}, stopChan chan struct{}, stoppedChan chan struct{}) {
|
|
go func() {
|
|
// Setup stoppedChan
|
|
defer close(stoppedChan)
|
|
|
|
utils.Logger().Debug().
|
|
Msg("Waiting for Consensus ready")
|
|
time.Sleep(30 * time.Second) // Wait for other nodes to be ready (test-only)
|
|
|
|
var newBlock *types.Block
|
|
|
|
// Set up the very first deadline.
|
|
deadline := time.Now().Add(node.BlockPeriod)
|
|
for {
|
|
// keep waiting for Consensus ready
|
|
select {
|
|
case <-stopChan:
|
|
utils.Logger().Debug().
|
|
Msg("Consensus new block proposal: STOPPED!")
|
|
return
|
|
case <-readySignal:
|
|
for {
|
|
time.Sleep(PeriodicBlock)
|
|
if time.Now().Before(deadline) {
|
|
continue
|
|
}
|
|
|
|
coinbase := node.Consensus.SelfAddress
|
|
if err := node.Worker.UpdateCurrent(coinbase); err != nil {
|
|
utils.Logger().Error().
|
|
Err(err).
|
|
Msg("Failed updating worker's state")
|
|
continue
|
|
}
|
|
|
|
// Normal tx block consensus
|
|
selectedTxs := node.getTransactionsForNewBlock(coinbase)
|
|
if err := node.Worker.UpdateCurrent(coinbase); err != nil {
|
|
utils.GetLogger().Error("Failed updating worker's state", "Error", err)
|
|
}
|
|
utils.Logger().Info().
|
|
Uint64("blockNum", node.Blockchain().CurrentBlock().NumberU64()+1).
|
|
Int("selectedTxs", len(selectedTxs)).
|
|
Msg("PROPOSING NEW BLOCK ------------------------------------------------")
|
|
|
|
if err := node.Worker.CommitTransactions(selectedTxs, coinbase); err != nil {
|
|
ctxerror.Log15(utils.GetLogger().Error,
|
|
ctxerror.New("cannot commit transactions").
|
|
WithCause(err))
|
|
}
|
|
sig, mask, err := node.Consensus.LastCommitSig()
|
|
if err != nil {
|
|
ctxerror.Log15(utils.GetLogger().Error,
|
|
ctxerror.New("Cannot got commit signatures from last block").
|
|
WithCause(err))
|
|
continue
|
|
}
|
|
|
|
// Propose cross shard receipts
|
|
receiptsList := node.proposeReceiptsProof()
|
|
if len(receiptsList) != 0 {
|
|
if err := node.Worker.CommitReceipts(receiptsList); err != nil {
|
|
ctxerror.Log15(utils.GetLogger().Error,
|
|
ctxerror.New("cannot commit receipts").
|
|
WithCause(err))
|
|
}
|
|
}
|
|
|
|
viewID := node.Consensus.GetViewID()
|
|
// add aggregated commit signatures from last block, except for the first two blocks
|
|
|
|
if node.NodeConfig.ShardID == 0 {
|
|
crossLinksToPropose, localErr := node.ProposeCrossLinkDataForBeaconchain()
|
|
if localErr == nil {
|
|
data, localErr := rlp.EncodeToBytes(crossLinksToPropose)
|
|
if localErr == nil {
|
|
newBlock, err = node.Worker.CommitWithCrossLinks(sig, mask, viewID, coinbase, data)
|
|
utils.Logger().Debug().
|
|
Uint64("blockNum", newBlock.NumberU64()).
|
|
Int("numCrossLinks", len(crossLinksToPropose)).
|
|
Msg("Successfully added cross links into new block")
|
|
} else {
|
|
utils.Logger().Debug().Err(localErr).Msg("ops0 ProposeCrossLinkDataForBeaconchain Failed")
|
|
}
|
|
} else {
|
|
utils.Logger().Debug().Err(localErr).Msg("ops1 ProposeCrossLinkDataForBeaconchain Failed")
|
|
newBlock, err = node.Worker.Commit(sig, mask, viewID, coinbase)
|
|
}
|
|
} else {
|
|
newBlock, err = node.Worker.Commit(sig, mask, viewID, coinbase)
|
|
}
|
|
|
|
if err != nil {
|
|
ctxerror.Log15(utils.GetLogger().Error,
|
|
ctxerror.New("cannot commit new block").
|
|
WithCause(err))
|
|
continue
|
|
} else if err := node.proposeShardStateWithoutBeaconSync(newBlock); err != nil {
|
|
ctxerror.Log15(utils.GetLogger().Error,
|
|
ctxerror.New("cannot add shard state").
|
|
WithCause(err))
|
|
} else {
|
|
utils.Logger().Debug().
|
|
Uint64("blockNum", newBlock.NumberU64()).
|
|
Int("numTxs", newBlock.Transactions().Len()).
|
|
Msg("Successfully proposed new block")
|
|
|
|
// Set deadline will be BlockPeriod from now at this place. Announce stage happens right after this.
|
|
deadline = time.Now().Add(node.BlockPeriod)
|
|
// Send the new block to Consensus so it can be confirmed.
|
|
node.BlockChannel <- newBlock
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (node *Node) proposeShardStateWithoutBeaconSync(block *types.Block) error {
|
|
if block == nil || !core.IsEpochLastBlock(block) {
|
|
return nil
|
|
}
|
|
nextEpoch := new(big.Int).Add(block.Header().Epoch(), common.Big1)
|
|
shardState := core.GetShardState(nextEpoch)
|
|
return block.AddShardState(shardState)
|
|
}
|
|
|
|
func (node *Node) proposeShardState(block *types.Block) error {
|
|
switch node.Consensus.ShardID {
|
|
case 0:
|
|
return node.proposeBeaconShardState(block)
|
|
default:
|
|
node.proposeLocalShardState(block)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (node *Node) proposeBeaconShardState(block *types.Block) error {
|
|
// TODO ek - replace this with variable epoch logic.
|
|
if !core.IsEpochLastBlock(block) {
|
|
// We haven't reached the end of this epoch; don't propose yet.
|
|
return nil
|
|
}
|
|
nextEpoch := new(big.Int).Add(block.Header().Epoch(), common.Big1)
|
|
shardState, err := core.CalculateNewShardState(
|
|
node.Blockchain(), nextEpoch, &node.CurrentStakes)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return block.AddShardState(shardState)
|
|
}
|
|
|
|
func (node *Node) proposeLocalShardState(block *types.Block) {
|
|
logger := block.Logger(utils.Logger())
|
|
// TODO ek – read this from beaconchain once BC sync is fixed
|
|
if node.nextShardState.master == nil {
|
|
logger.Debug().Msg("yet to receive master proposal from beaconchain")
|
|
return
|
|
}
|
|
|
|
nlogger := logger.With().
|
|
Uint64("nextEpoch", node.nextShardState.master.Epoch).
|
|
Time("proposeTime", node.nextShardState.proposeTime).
|
|
Logger()
|
|
logger = &nlogger
|
|
if time.Now().Before(node.nextShardState.proposeTime) {
|
|
logger.Debug().Msg("still waiting for shard state to propagate")
|
|
return
|
|
}
|
|
masterShardState := node.nextShardState.master.ShardState
|
|
var localShardState shard.State
|
|
committee := masterShardState.FindCommitteeByID(block.ShardID())
|
|
if committee != nil {
|
|
logger.Info().Msg("found local shard info; proposing it")
|
|
localShardState = append(localShardState, *committee)
|
|
} else {
|
|
logger.Info().Msg("beacon committee disowned us; proposing nothing")
|
|
// Leave local proposal empty to signal the end of shard (disbanding).
|
|
}
|
|
err := block.AddShardState(localShardState)
|
|
if err != nil {
|
|
logger.Error().Err(err).Msg("Failed proposin local shard state")
|
|
}
|
|
}
|
|
|
|
func (node *Node) proposeReceiptsProof() []*types.CXReceiptsProof {
|
|
validReceiptsList := []*types.CXReceiptsProof{}
|
|
pendingReceiptsList := []*types.CXReceiptsProof{}
|
|
node.pendingCXMutex.Lock()
|
|
|
|
sort.Slice(node.pendingCXReceipts, func(i, j int) bool {
|
|
return node.pendingCXReceipts[i].MerkleProof.ShardID < node.pendingCXReceipts[j].MerkleProof.ShardID || (node.pendingCXReceipts[i].MerkleProof.ShardID == node.pendingCXReceipts[j].MerkleProof.ShardID && node.pendingCXReceipts[i].MerkleProof.BlockNum.Cmp(node.pendingCXReceipts[j].MerkleProof.BlockNum) < 0)
|
|
})
|
|
|
|
m := make(map[common.Hash]bool)
|
|
|
|
for _, cxp := range node.pendingCXReceipts {
|
|
// 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] = true
|
|
}
|
|
|
|
if err := node.compareCrosslinkWithReceipts(cxp); err != nil {
|
|
utils.Logger().Debug().Err(err).Interface("cxp", cxp).Msg("[proposeReceiptsProof] CrossLink Verify Fail")
|
|
if err != ErrCrosslinkVerificationFail {
|
|
pendingReceiptsList = append(pendingReceiptsList, cxp)
|
|
}
|
|
} else {
|
|
utils.Logger().Debug().Interface("cxp", cxp).Msg("[proposeReceiptsProof] CXReceipts Added")
|
|
validReceiptsList = append(validReceiptsList, cxp)
|
|
}
|
|
}
|
|
node.pendingCXReceipts = pendingReceiptsList
|
|
node.pendingCXMutex.Unlock()
|
|
utils.Logger().Debug().Msgf("[proposeReceiptsProof] number of validReceipts %d, pendingReceipts %d", len(validReceiptsList), len(pendingReceiptsList))
|
|
return validReceiptsList
|
|
}
|
|
|