Ungated fixes extracted from harmony-one/harmony#5122

consensus/checks.go

@@ -187,6 +187,18 @@ func (consensus *Consensus) onViewChangeSanityCheck(recvMsg *FBFTMessage) bool {
 	}
 	senderKey := recvMsg.SenderPubkeys[0]
 
+	// The sender's signature is only meaningful as a vote if the sender is in the
+	// committee voting. Signatures collected here are aggregated into the new view
+	// message while the accompanying bitmap can only describe committee members,
+	// so a signature from outside the committee is one the bitmap cannot account
+	// for and the aggregate would no longer match it.
+	if consensus.decider().IndexOf(senderKey.Bytes) == -1 {
+		consensus.getLogger().Warn().
+			Str("sender", senderKey.Bytes.Hex()).
+			Msg("[onViewChangeSanityCheck] sender is not in the committee")
+		return false
+	}
+
 	viewIDHash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(viewIDHash, recvMsg.ViewID)
 	if !recvMsg.ViewidSig.VerifyHash(senderKey.Object, viewIDHash) {

consensus/consensus.go

@@ -78,6 +78,10 @@ type Consensus struct {
 	multiSigBitmap *bls_cosi.Mask // Bitmap for parsing multisig bitmap from validators
 
 	pendingCXReceipts map[utils.CXKey]*types.CXReceiptsProof // All the receipts received but not yet processed for Consensus
+	// Number of proposal rounds each pending receipt has been retried while its
+	// source shard state was still unavailable, so the pending set keeps turning
+	// over rather than accumulating receipts that never become verifiable.
+	pendingCXReceiptsDeferrals map[utils.CXKey]int
 	// Registry for services.
 	registry *registry.Registry
 	// Minimal number of peers in the shard
@@ -299,9 +303,10 @@ func New(
 		host:         host,
 		msgSender:    NewMessageSender(host),
 		// FBFT timeout
-		consensusTimeout:  createTimeout(),
-		dHelper:           downloadAsync{},
-		pendingCXReceipts: make(map[utils.CXKey]*types.CXReceiptsProof), // All the receipts received but not yet processed for Consensus
+		consensusTimeout:           createTimeout(),
+		dHelper:                    downloadAsync{},
+		pendingCXReceipts:          make(map[utils.CXKey]*types.CXReceiptsProof), // All the receipts received but not yet processed for Consensus
+		pendingCXReceiptsDeferrals: make(map[utils.CXKey]int),
 	}
 	registry.SetQuorum(Decider)
 

consensus/consensus_block_proposing.go

@@ -1,6 +1,7 @@
 package consensus
 
 import (
+	"math/big"
 	"sort"
 	"strings"
 	"time"
@@ -19,6 +20,9 @@ import (
 const (
 	IncomingReceiptsLimit = 2000 // 2000 * (numShards - 1)
 	SleepPeriod           = 20 * time.Millisecond
+	// maxPendingCXReceiptDeferrals is how many proposal rounds a pending
+	// CXReceiptsProof may stay unverifiable before it is dropped.
+	maxPendingCXReceiptDeferrals = 10
 )
 
 // ProposeNewBlock proposes a new block...
@@ -283,6 +287,7 @@ func (consensus *Consensus) proposeReceiptsProof() []*types.CXReceiptsProof {
 	})
 
 	m := map[common.Hash]struct{}{}
+	deferrals := map[utils.CXKey]int{}
 
 Loop:
 	for _, cxp := range consensus.pendingCXReceipts {
@@ -310,7 +315,14 @@ Loop:
 		}
 
 		if err := core.NewBlockValidator(consensus.Blockchain()).ValidateCXReceiptsProof(cxp); err != nil {
-			if strings.Contains(err.Error(), rawdb.MsgNoShardStateFromDB) {
+			// A missing source shard state means the commit signature has not been
+			// checked yet rather than found wrong, so the proof is worth retrying on
+			// a later round. Retries are capped so the pending set keeps turning over
+			// when a shard state never arrives.
+			key := utils.GetPendingCXKey(cxp.Header.ShardID(), cxp.Header.Number().Uint64())
+			if strings.Contains(err.Error(), rawdb.MsgNoShardStateFromDB) &&
+				consensus.pendingCXReceiptsDeferrals[key] < maxPendingCXReceiptDeferrals {
+				deferrals[key] = consensus.pendingCXReceiptsDeferrals[key] + 1
 				pendingReceiptsList = append(pendingReceiptsList, cxp)
 			} else {
 				consensus.getLogger().Error().Err(err).Msg("[proposeReceiptsProof] Invalid CXReceiptsProof")
@@ -330,6 +342,9 @@ Loop:
 		key := utils.GetPendingCXKey(shardID, blockNum)
 		consensus.pendingCXReceipts[key] = v
 	}
+	// Deferral counts only survive for receipts still pending; entries dropped
+	// above lose their count along with their slot.
+	consensus.pendingCXReceiptsDeferrals = deferrals
 
 	consensus.getLogger().Debug().Msgf("[proposeReceiptsProof] number of validReceipts %d", len(validReceiptsList))
 	return validReceiptsList
@@ -350,6 +365,21 @@ func (consensus *Consensus) AddPendingReceipts(receipts *types.CXReceiptsProof)
 
 	// Sanity checks
 
+	// Validation below verifies the header commit signature against the source
+	// shard state of the header's epoch, and tolerates that shard state being
+	// absent so receipts arriving just before an epoch transition are still kept.
+	// That tolerance is meant for the epoch we are about to enter, so the epoch a
+	// receipt claims is only meaningful up to one past the current one.
+	curEpoch := consensus.Blockchain().CurrentHeader().Epoch()
+	maxEpoch := new(big.Int).Add(curEpoch, common.Big1)
+	if e := receipts.Header.Epoch(); e == nil || e.Cmp(maxEpoch) > 0 {
+		consensus.getLogger().Info().
+			Interface("incoming-epoch", e).
+			Str("max-accepted-epoch", maxEpoch.String()).
+			Msg("[AddPendingReceipts] Incoming receipt epoch too far ahead")
+		return
+	}
+
 	if err := core.NewBlockValidator(consensus.Blockchain()).ValidateCXReceiptsProof(receipts); err != nil {
 		if !strings.Contains(err.Error(), rawdb.MsgNoShardStateFromDB) {
 			consensus.getLogger().Error().Err(err).Msg("[AddPendingReceipts] Invalid CXReceiptsProof")

consensus/pending_cx_receipts_test.go

@@ -0,0 +1,184 @@
+package consensus
+
+import (
+	"bytes"
+	"encoding/binary"
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	blockfactory "github.com/harmony-one/harmony/block/factory"
+	"github.com/harmony-one/harmony/consensus/quorum"
+	corepkg "github.com/harmony-one/harmony/core"
+	"github.com/harmony-one/harmony/core/rawdb"
+	coretypes "github.com/harmony-one/harmony/core/types"
+	"github.com/harmony-one/harmony/core/vm"
+	hmybls "github.com/harmony-one/harmony/crypto/bls"
+	chain2 "github.com/harmony-one/harmony/internal/chain"
+	"github.com/harmony-one/harmony/internal/params"
+	"github.com/harmony-one/harmony/internal/registry"
+	"github.com/harmony-one/harmony/multibls"
+	workerpkg "github.com/harmony-one/harmony/node/harmony/worker"
+	"github.com/harmony-one/harmony/shard"
+	"github.com/stretchr/testify/require"
+)
+
+// pendingCXReceiptsConfig is the mainnet config - which the block proposal path
+// is known to work under - with the cross-shard forks moved to epoch 0 so the
+// receipt paths are live from genesis.
+func pendingCXReceiptsConfig() *params.ChainConfig {
+	cfg := *params.MainnetChainConfig
+	cfg.CrossTxEpoch = big.NewInt(0)
+	cfg.CXMerkleProofReplayFixEpoch = big.NewInt(0)
+	return &cfg
+}
+
+func pendingCXReceiptsChain(
+	t *testing.T, shardID uint32, beacon corepkg.BlockChain, st shard.State,
+) *corepkg.BlockChainImpl {
+	t.Helper()
+
+	db := rawdb.NewMemoryDatabase()
+	gspec := corepkg.Genesis{
+		Config:     pendingCXReceiptsConfig(),
+		Factory:    blockfactory.ForMainnet,
+		Alloc:      corepkg.GenesisAlloc{},
+		ShardID:    shardID,
+		GasLimit:   params.TestGenesisGasLimit,
+		ShardState: st,
+	}
+	gspec.MustCommit(db)
+
+	chain, err := corepkg.NewBlockChain(
+		db, nil, beacon,
+		&corepkg.CacheConfig{SnapshotLimit: 0},
+		gspec.Config, chain2.NewEngine(), vm.Config{},
+	)
+	require.NoError(t, err)
+	return chain
+}
+
+// pendingCXReceiptsHarness builds a shard-1 consensus backed by a real chain so
+// that AddPendingReceipts can consult CurrentHeader/Config. The chain is grown
+// past block 1 because header signature verification is skipped below that
+// height, which would make every proof validate regardless of its signature.
+func pendingCXReceiptsHarness(t *testing.T) *Consensus {
+	t.Helper()
+
+	beaconSigner := hmybls.RandPrivateKey()
+	beaconPub := hmybls.PublicKeyWrapper{Object: beaconSigner.GetPublicKey()}
+	require.NoError(t, beaconPub.Bytes.FromLibBLSPublicKey(beaconPub.Object))
+	shardSigner := hmybls.RandPrivateKey()
+	shardPub := hmybls.PublicKeyWrapper{Object: shardSigner.GetPublicKey()}
+	require.NoError(t, shardPub.Bytes.FromLibBLSPublicKey(shardPub.Object))
+
+	st := crossLinkCacheShardState(
+		common.Big0,
+		common.BytesToAddress([]byte{0xb0}),
+		common.BytesToAddress([]byte{0x51}),
+		beaconPub.Bytes,
+		shardPub.Bytes,
+	)
+	beaconChain := pendingCXReceiptsChain(t, 0, nil, st)
+	shardChain := pendingCXReceiptsChain(t, 1, beaconChain, st)
+
+	txPoolConfig := corepkg.DefaultTxPoolConfig
+	txPoolConfig.Journal = ""
+	txPool := corepkg.NewTxPool(
+		txPoolConfig, pendingCXReceiptsConfig(), shardChain,
+		coretypes.NewTransactionErrorSink(),
+	)
+	t.Cleanup(txPool.Stop)
+
+	reg := registry.New().
+		SetBlockchain(shardChain).
+		SetBeaconchain(beaconChain).
+		SetTxPool(txPool).
+		SetCxPool(corepkg.NewCxPool(corepkg.CxPoolSize)).
+		SetWorker(workerpkg.New(shardChain, beaconChain)).
+		SetAddressToBLSKey(crossLinkCacheAddressToBLSKey{shardID: shardChain.ShardID()})
+
+	decider := quorum.NewDecider(quorum.SuperMajorityStake, shardChain.ShardID())
+	consensus, err := New(
+		nil, shardChain.ShardID(), multibls.GetPrivateKeys(shardSigner),
+		reg, decider, 1, false,
+	)
+	require.NoError(t, err)
+	consensus.SetLeaderPubKey(&shardPub)
+	consensus.SetViewIDs(shardChain.CurrentBlock().NumberU64())
+
+	signer := hmybls.PrivateKeyWrapper{Pri: shardSigner, Pub: &shardPub}
+	var lastCommitSig []byte
+	for shardChain.CurrentBlock().NumberU64() < 2 {
+		blk := crossLinkCacheProposeAndInsertBeaconBlock(
+			t, consensus, shardChain, signer, lastCommitSig, nil,
+		)
+		lastCommitSig = blk.GetCurrentCommitSig()
+	}
+	require.Greater(t, shardChain.CurrentHeader().Number().Uint64(), uint64(1))
+	return consensus
+}
+
+// makeCXProof builds a CXReceiptsProof whose merkle proof and header agree with
+// each other. Every field here is derived locally, so the proof is internally
+// consistent but carries no commit signature from the source shard committee.
+func makeCXProof(t *testing.T, srcShard uint32, blockNum uint64, epoch *big.Int, toShard uint32) *coretypes.CXReceiptsProof {
+	t.Helper()
+
+	to := common.BytesToAddress([]byte{0x42})
+	receipts := coretypes.CXReceipts{{
+		TxHash:    common.Hash{0x01},
+		From:      common.BytesToAddress([]byte{0x11}),
+		To:        &to,
+		ShardID:   srcShard,
+		ToShardID: toShard,
+		Amount:    big.NewInt(1),
+	}}
+	shardHash := coretypes.DeriveSha(receipts)
+
+	proof := &coretypes.CXMerkleProof{
+		BlockNum:      new(big.Int).SetUint64(blockNum),
+		ShardID:       srcShard,
+		ShardIDs:      []uint32{toShard},
+		CXShardHashes: []common.Hash{shardHash},
+	}
+
+	// Same derivation ValidateCXReceiptsProof performs over the merkle proof.
+	buf := bytes.Buffer{}
+	for j := range proof.ShardIDs {
+		sKey := make([]byte, 4)
+		binary.BigEndian.PutUint32(sKey, proof.ShardIDs[j])
+		buf.Write(sKey)
+		buf.Write(proof.CXShardHashes[j][:])
+	}
+
+	header := blockfactory.ForMainnet.NewHeader(epoch)
+	header.SetNumber(new(big.Int).SetUint64(blockNum))
+	header.SetShardID(srcShard)
+	header.SetOutgoingReceiptHash(crypto.Keccak256Hash(buf.Bytes()))
+	proof.CXReceiptHash = header.OutgoingReceiptHash()
+	proof.BlockHash = header.Hash()
+
+	return &coretypes.CXReceiptsProof{
+		Receipts:     receipts,
+		MerkleProof:  proof,
+		Header:       header,
+		CommitSig:    make([]byte, 96),
+		CommitBitmap: []byte{0x01},
+	}
+}
+
+// TestAddPendingReceiptsRejectsFutureEpoch checks that a receipts proof claiming
+// an epoch far beyond the current one is not admitted to the pending pool. Only
+// the epoch the chain is about to enter can legitimately lack a shard state.
+func TestAddPendingReceiptsRejectsFutureEpoch(t *testing.T) {
+	consensus := pendingCXReceiptsHarness(t)
+	myShard := consensus.Blockchain().ShardID()
+	curEpoch := consensus.Blockchain().CurrentHeader().Epoch()
+
+	farFuture := new(big.Int).Add(curEpoch, big.NewInt(1000))
+	consensus.AddPendingReceipts(makeCXProof(t, 0, 7, farFuture, myShard))
+	require.Empty(t, consensus.PendingCXReceipts(),
+		"a proof claiming a far future epoch should not be pending")
+}