Merge pull request #4470 from harmony-one/dev

Release Candidate 2023.2.5 (dev -> main)

api/service/stagedstreamsync/adapter.go

@@ -18,6 +18,8 @@ type syncProtocol interface {
 	GetRawBlocksByNumber(ctx context.Context, bns []uint64, opts ...syncproto.Option) ([][]byte, [][]byte, sttypes.StreamID, error)
 	GetBlockHashes(ctx context.Context, bns []uint64, opts ...syncproto.Option) ([]common.Hash, sttypes.StreamID, error)
 	GetBlocksByHashes(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) ([]*types.Block, sttypes.StreamID, error)
+	GetReceipts(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) (receipts []types.Receipts, stid sttypes.StreamID, err error)
+	GetNodeData(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) (data [][]byte, stid sttypes.StreamID, err error)
 
 	RemoveStream(stID sttypes.StreamID) // If a stream delivers invalid data, remove the stream
 	StreamFailed(stID sttypes.StreamID, reason string)

consensus/checks.go

@@ -68,7 +68,7 @@ func (consensus *Consensus) isRightBlockNumAndViewID(recvMsg *FBFTMessage) bool
 }
 
 func (consensus *Consensus) onAnnounceSanityChecks(recvMsg *FBFTMessage) bool {
-	logMsgs := consensus.FBFTLog.GetMessagesByTypeSeqView(
+	logMsgs := consensus.fBFTLog.GetMessagesByTypeSeqView(
 		msg_pb.MessageType_ANNOUNCE, recvMsg.BlockNum, recvMsg.ViewID,
 	)
 	if len(logMsgs) > 0 {

consensus/consensus.go

@@ -46,7 +46,7 @@ type VerifyBlockFunc func(*types.Block) error
 type Consensus struct {
 	Decider quorum.Decider
 	// FBFTLog stores the pbft messages and blocks during FBFT process
-	FBFTLog *FBFTLog
+	fBFTLog *FBFTLog
 	// phase: different phase of FBFT protocol: pre-prepare, prepare, commit, finish etc
 	phase FBFTPhase
 	// current indicates what state a node is in
@@ -84,7 +84,7 @@ type Consensus struct {
 	// IgnoreViewIDCheck determines whether to ignore viewID check
 	IgnoreViewIDCheck *abool.AtomicBool
 	// consensus mutex
-	mutex sync.RWMutex
+	mutex *sync.RWMutex
 	// ViewChange struct
 	vc *viewChange
 	// Signal channel for proposing a new block and start new consensus
@@ -140,6 +140,13 @@ func (consensus *Consensus) Blockchain() core.BlockChain {
 	return consensus.registry.GetBlockchain()
 }
 
+func (consensus *Consensus) FBFTLog() FBFT {
+	return threadsafeFBFTLog{
+		log: consensus.fBFTLog,
+		mu:  consensus.mutex,
+	}
+}
+
 // ChainReader returns the chain reader.
 // This is mostly the same as Blockchain, but it returns only read methods, so we assume it's safe for concurrent use.
 func (consensus *Consensus) ChainReader() engine.ChainReader {
@@ -165,11 +172,11 @@ func (consensus *Consensus) Beaconchain() core.BlockChain {
 
 // VerifyBlock is a function used to verify the block and keep trace of verified blocks.
 func (consensus *Consensus) verifyBlock(block *types.Block) error {
-	if !consensus.FBFTLog.IsBlockVerified(block.Hash()) {
+	if !consensus.fBFTLog.IsBlockVerified(block.Hash()) {
 		if err := consensus.BlockVerifier(block); err != nil {
 			return errors.Errorf("Block verification failed: %s", err)
 		}
-		consensus.FBFTLog.MarkBlockVerified(block)
+		consensus.fBFTLog.MarkBlockVerified(block)
 	}
 	return nil
 }
@@ -261,21 +268,21 @@ func New(
 	Decider quorum.Decider, minPeers int, aggregateSig bool,
 ) (*Consensus, error) {
 	consensus := Consensus{
-		ShardID: shard,
+		mutex:           &sync.RWMutex{},
+		ShardID:         shard,
+		fBFTLog:         NewFBFTLog(),
+		phase:           FBFTAnnounce,
+		current:         State{mode: Normal},
+		Decider:         Decider,
+		registry:        registry,
+		MinPeers:        minPeers,
+		AggregateSig:    aggregateSig,
+		host:            host,
+		msgSender:       NewMessageSender(host),
+		BlockNumLowChan: make(chan struct{}, 1),
+		// FBFT timeout
+		consensusTimeout: createTimeout(),
 	}
-	consensus.Decider = Decider
-	consensus.registry = registry
-	consensus.MinPeers = minPeers
-	consensus.AggregateSig = aggregateSig
-	consensus.host = host
-	consensus.msgSender = NewMessageSender(host)
-	consensus.BlockNumLowChan = make(chan struct{}, 1)
-	// FBFT related
-	consensus.FBFTLog = NewFBFTLog()
-	consensus.phase = FBFTAnnounce
-	consensus.current = State{mode: Normal}
-	// FBFT timeout
-	consensus.consensusTimeout = createTimeout()
 
 	if multiBLSPriKey != nil {
 		consensus.priKey = multiBLSPriKey

consensus/consensus_service.go

@@ -145,9 +145,9 @@ func (consensus *Consensus) updateBitmaps() {
 		Str("MessageType", consensus.phase.String()).
 		Msg("[UpdateBitmaps] Updating consensus bitmaps")
 	members := consensus.Decider.Participants()
-	prepareBitmap, _ := bls_cosi.NewMask(members, nil)
-	commitBitmap, _ := bls_cosi.NewMask(members, nil)
-	multiSigBitmap, _ := bls_cosi.NewMask(members, nil)
+	prepareBitmap := bls_cosi.NewMask(members)
+	commitBitmap := bls_cosi.NewMask(members)
+	multiSigBitmap := bls_cosi.NewMask(members)
 	consensus.prepareBitmap = prepareBitmap
 	consensus.commitBitmap = commitBitmap
 	consensus.multiSigBitmap = multiSigBitmap
@@ -575,7 +575,7 @@ func (consensus *Consensus) selfCommit(payload []byte) error {
 	copy(blockHash[:], payload[:32])
 
 	// Leader sign and add commit message
-	block := consensus.FBFTLog.GetBlockByHash(blockHash)
+	block := consensus.fBFTLog.GetBlockByHash(blockHash)
 	if block == nil {
 		return errGetPreparedBlock
 	}
@@ -627,11 +627,8 @@ func (consensus *Consensus) NumSignaturesIncludedInBlock(block *types.Block) uin
 	count := uint32(0)
 	members := consensus.Decider.Participants()
 	// TODO(audit): do not reconstruct the Mask
-	mask, err := bls.NewMask(members, nil)
-	if err != nil {
-		return count
-	}
-	err = mask.SetMask(block.Header().LastCommitBitmap())
+	mask := bls.NewMask(members)
+	err := mask.SetMask(block.Header().LastCommitBitmap())
 	if err != nil {
 		return count
 	}

consensus/consensus_test.go

@@ -22,7 +22,6 @@ func TestConsensusInitialization(t *testing.T) {
 	assert.NoError(t, err)
 
 	messageSender := &MessageSender{host: host, retryTimes: int(phaseDuration.Seconds()) / RetryIntervalInSec}
-	fbtLog := NewFBFTLog()
 	state := State{mode: Normal}
 
 	timeouts := createTimeout()
@@ -37,7 +36,7 @@ func TestConsensusInitialization(t *testing.T) {
 	assert.IsType(t, make(chan struct{}), consensus.BlockNumLowChan)
 
 	// FBFTLog
-	assert.Equal(t, fbtLog, consensus.FBFTLog)
+	assert.NotNil(t, consensus.FBFTLog())
 
 	assert.Equal(t, FBFTAnnounce, consensus.phase)
 

consensus/consensus_v2.go

@@ -161,10 +161,10 @@ func (consensus *Consensus) finalCommit() {
 		network.Bytes,
 		network.FBFTMsg
 	commitSigAndBitmap := FBFTMsg.Payload
-	consensus.FBFTLog.AddVerifiedMessage(FBFTMsg)
+	consensus.fBFTLog.AddVerifiedMessage(FBFTMsg)
 	// find correct block content
 	curBlockHash := consensus.blockHash
-	block := consensus.FBFTLog.GetBlockByHash(curBlockHash)
+	block := consensus.fBFTLog.GetBlockByHash(curBlockHash)
 	if block == nil {
 		consensus.getLogger().Warn().
 			Str("blockHash", hex.EncodeToString(curBlockHash[:])).
@@ -278,7 +278,7 @@ func (consensus *Consensus) BlockCommitSigs(blockNum uint64) ([]byte, error) {
 	lastCommits, err := consensus.Blockchain().ReadCommitSig(blockNum)
 	if err != nil ||
 		len(lastCommits) < bls.BLSSignatureSizeInBytes {
-		msgs := consensus.FBFTLog.GetMessagesByTypeSeq(
+		msgs := consensus.FBFTLog().GetMessagesByTypeSeq(
 			msg_pb.MessageType_COMMITTED, blockNum,
 		)
 		if len(msgs) != 1 {
@@ -482,7 +482,7 @@ func (consensus *Consensus) getLastMileBlockIter(bnStart uint64, cb func(iter *L
 	}
 	return cb(&LastMileBlockIter{
 		blockCandidates: blocks,
-		fbftLog:         consensus.FBFTLog,
+		fbftLog:         consensus.fBFTLog,
 		verify:          consensus.BlockVerifier,
 		curIndex:        0,
 		logger:          consensus.getLogger(),
@@ -513,7 +513,7 @@ func (consensus *Consensus) getLastMileBlocksAndMsg(bnStart uint64) ([]*types.Bl
 		msgs   []*FBFTMessage
 	)
 	for blockNum := bnStart; ; blockNum++ {
-		blk, msg, err := consensus.FBFTLog.GetCommittedBlockAndMsgsFromNumber(blockNum, consensus.getLogger())
+		blk, msg, err := consensus.fBFTLog.GetCommittedBlockAndMsgsFromNumber(blockNum, consensus.getLogger())
 		if err != nil {
 			if err == errFBFTLogNotFound {
 				break
@@ -551,7 +551,7 @@ func (consensus *Consensus) preCommitAndPropose(blk *types.Block) error {
 		network.Bytes,
 		network.FBFTMsg
 	bareMinimumCommit := FBFTMsg.Payload
-	consensus.FBFTLog.AddVerifiedMessage(FBFTMsg)
+	consensus.fBFTLog.AddVerifiedMessage(FBFTMsg)
 
 	if err := consensus.verifyLastCommitSig(bareMinimumCommit, blk); err != nil {
 		return errors.Wrap(err, "[preCommitAndPropose] failed verifying last commit sig")
@@ -567,16 +567,16 @@ func (consensus *Consensus) preCommitAndPropose(blk *types.Block) error {
 				nodeconfig.NewGroupIDByShardID(nodeconfig.ShardID(consensus.ShardID)),
 			},
 			p2p.ConstructMessage(msgToSend)); err != nil {
-			consensus.getLogger().Warn().Err(err).Msg("[preCommitAndPropose] Cannot send committed message")
+			consensus.GetLogger().Warn().Err(err).Msg("[preCommitAndPropose] Cannot send committed message")
 		} else {
-			consensus.getLogger().Info().
+			consensus.GetLogger().Info().
 				Str("blockHash", blk.Hash().Hex()).
 				Uint64("blockNum", consensus.BlockNum()).
 				Hex("lastCommitSig", bareMinimumCommit).
 				Msg("[preCommitAndPropose] Sent Committed Message")
 		}
 
-		if _, err := consensus.Blockchain().InsertChain([]*types.Block{blk}, !consensus.FBFTLog.IsBlockVerified(blk.Hash())); err != nil {
+		if _, err := consensus.Blockchain().InsertChain([]*types.Block{blk}, !consensus.FBFTLog().IsBlockVerified(blk.Hash())); err != nil {
 			consensus.getLogger().Error().Err(err).Msg("[preCommitAndPropose] Failed to add block to chain")
 			return
 		}
@@ -661,7 +661,7 @@ func (consensus *Consensus) tryCatchup() error {
 
 func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMessage) error {
 	if consensus.Blockchain().CurrentBlock().NumberU64() < blk.NumberU64() {
-		if _, err := consensus.Blockchain().InsertChain([]*types.Block{blk}, !consensus.FBFTLog.IsBlockVerified(blk.Hash())); err != nil {
+		if _, err := consensus.Blockchain().InsertChain([]*types.Block{blk}, !consensus.fBFTLog.IsBlockVerified(blk.Hash())); err != nil {
 			consensus.getLogger().Error().Err(err).Msg("[commitBlock] Failed to add block to chain")
 			return err
 		}
@@ -785,7 +785,7 @@ func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg
 	if blk.IsLastBlockInEpoch() {
 		consensus.setMode(consensus.updateConsensusInformation())
 	}
-	consensus.FBFTLog.PruneCacheBeforeBlock(blk.NumberU64())
+	consensus.fBFTLog.PruneCacheBeforeBlock(blk.NumberU64())
 	consensus.resetState()
 }
 
@@ -920,19 +920,12 @@ func (consensus *Consensus) ValidateVdfAndProof(headerObj *block.Header) bool {
 func (consensus *Consensus) DeleteBlocksLessThan(number uint64) {
 	consensus.mutex.Lock()
 	defer consensus.mutex.Unlock()
-	consensus.FBFTLog.deleteBlocksLessThan(number)
+	consensus.fBFTLog.deleteBlocksLessThan(number)
 }
 
 // DeleteMessagesLessThan deletes messages less than given block number.
 func (consensus *Consensus) DeleteMessagesLessThan(number uint64) {
 	consensus.mutex.Lock()
 	defer consensus.mutex.Unlock()
-	consensus.FBFTLog.deleteMessagesLessThan(number)
-}
-
-// DeleteBlockByNumber deletes block by given block number.
-func (consensus *Consensus) DeleteBlockByNumber(number uint64) {
-	consensus.mutex.Lock()
-	defer consensus.mutex.Unlock()
-	consensus.FBFTLog.deleteBlockByNumber(number)
+	consensus.fBFTLog.deleteMessagesLessThan(number)
 }

consensus/construct.go

@@ -82,11 +82,7 @@ func (consensus *Consensus) construct(
 		)
 	} else {
 		// TODO: use a persistent bitmap to report bitmap
-		mask, err := bls.NewMask(consensus.Decider.Participants(), nil)
-		if err != nil {
-			utils.Logger().Warn().Err(err).Msg("unable to setup mask for multi-sig message")
-			return nil, err
-		}
+		mask := bls.NewMask(consensus.Decider.Participants())
 		for _, key := range priKeys {
 			mask.SetKey(key.Pub.Bytes, true)
 		}

consensus/double_sign.go

@@ -22,7 +22,7 @@ func (consensus *Consensus) checkDoubleSign(recvMsg *FBFTMessage) bool {
 			); alreadyCastBallot != nil {
 				for _, pubKey1 := range alreadyCastBallot.SignerPubKeys {
 					if bytes.Compare(pubKey2.Bytes[:], pubKey1[:]) == 0 {
-						for _, blk := range consensus.FBFTLog.GetBlocksByNumber(recvMsg.BlockNum) {
+						for _, blk := range consensus.fBFTLog.GetBlocksByNumber(recvMsg.BlockNum) {
 							firstSignedHeader := blk.Header()
 							areHeightsEqual := firstSignedHeader.Number().Uint64() == recvMsg.BlockNum
 							areViewIDsEqual := firstSignedHeader.ViewID().Uint64() == recvMsg.ViewID
@@ -138,8 +138,8 @@ func (consensus *Consensus) couldThisBeADoubleSigner(
 	recvMsg *FBFTMessage,
 ) bool {
 	num, hash := consensus.BlockNum(), recvMsg.BlockHash
-	suspicious := !consensus.FBFTLog.HasMatchingAnnounce(num, hash) ||
-		!consensus.FBFTLog.HasMatchingPrepared(num, hash)
+	suspicious := !consensus.fBFTLog.HasMatchingAnnounce(num, hash) ||
+		!consensus.fBFTLog.HasMatchingPrepared(num, hash)
 	if suspicious {
 		consensus.getLogger().Debug().
 			Str("message", recvMsg.String()).

consensus/engine/consensus_engine.go

@@ -23,6 +23,27 @@ type ChainReader interface {
 	// Config retrieves the blockchain's chain configuration.
 	Config() *params.ChainConfig
 
+	// TrieNode retrieves a blob of data associated with a trie node
+	// either from ephemeral in-memory cache, or from persistent storage.
+	TrieNode(hash common.Hash) ([]byte, error)
+
+	// ContractCode retrieves a blob of data associated with a contract
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ContractCode(hash common.Hash) ([]byte, error)
+
+	// ValidatorCode retrieves a blob of data associated with a validator
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ValidatorCode(hash common.Hash) ([]byte, error)
+
+	// GetReceiptsByHash retrieves the receipts for all transactions in a given block.
+	GetReceiptsByHash(hash common.Hash) types.Receipts
+
 	// CurrentHeader retrieves the current header from the local chain.
 	CurrentHeader() *block.Header