prototype: compact blocks

config/config.go

@@ -692,9 +692,7 @@ func DefaultFuzzConnConfig() *FuzzConnConfig {
 // MempoolConfig defines the configuration options for the CometBFT mempool
 type MempoolConfig struct {
 	// Mempool version to use:
-	//  1) "v0" - FIFO mempool.
-	//  2) "v1" - (default) prioritized mempool.
-	//  3) "v2" - content addressable transaction pool
+	//  1) "v2" - (default) content addressable transaction pool
 	Version string `mapstructure:"version"`
 	// RootDir is the root directory for all data. This should be configured via
 	// the $CMTHOME env variable or --home cmd flag rather than overriding this
@@ -763,7 +761,7 @@ type MempoolConfig struct {
 // DefaultMempoolConfig returns a default configuration for the CometBFT mempool
 func DefaultMempoolConfig() *MempoolConfig {
 	return &MempoolConfig{
-		Version:   MempoolV1,
+		Version:   MempoolV2,
 		Recheck:   true,
 		Broadcast: true,
 		WalPath:   "",
@@ -798,6 +796,9 @@ func (cfg *MempoolConfig) WalEnabled() bool {
 // ValidateBasic performs basic validation (checking param bounds, etc.) and
 // returns an error if any check fails.
 func (cfg *MempoolConfig) ValidateBasic() error {
+	if cfg.Version != MempoolV2 {
+		return errors.New("only v2 mempool is supported for compact blocks")
+	}
 	if cfg.Size < 0 {
 		return errors.New("size can't be negative")
 	}

config/toml.go

@@ -343,9 +343,7 @@ dial_timeout = "{{ .P2P.DialTimeout }}"
 [mempool]
 
 # Mempool version to use:
-#   1) "v0" - FIFO mempool.
-#   2) "v1" - (default) prioritized mempool.
-#   3) "v2" - content addressable transaction pool
+#   1) "v2" - (default) content addressable transaction pool
 version = "{{ .Mempool.Version }}"
 
 # Recheck (default: true) defines whether CometBFT should recheck the

consensus/byzantine_test.go

@@ -20,12 +20,8 @@ import (
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
 	cmtsync "github.com/tendermint/tendermint/libs/sync"
-	mempl "github.com/tendermint/tendermint/mempool"
 
-	cfg "github.com/tendermint/tendermint/config"
-	mempoolv2 "github.com/tendermint/tendermint/mempool/cat"
-	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
-	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
+	"github.com/tendermint/tendermint/mempool/cat"
 	"github.com/tendermint/tendermint/p2p"
 	cmtcons "github.com/tendermint/tendermint/proto/tendermint/consensus"
 	cmtproto "github.com/tendermint/tendermint/proto/tendermint/types"
@@ -48,6 +44,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 	genDoc, privVals := randGenesisDoc(nValidators, false, 30)
 	css := make([]*State, nValidators)
+	catReactors := make([]*cat.Reactor, nValidators)
 
 	for i := 0; i < nValidators; i++ {
 		logger := consensusLogger().With("test", "byzantine", "validator", i)
@@ -72,33 +69,21 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		proxyAppConnConMem := abcicli.NewLocalClient(mtx, app)
 
 		// Make Mempool
-		var mempool mempl.Mempool
-
-		switch thisConfig.Mempool.Version {
-		case cfg.MempoolV0:
-			mempool = mempoolv0.NewCListMempool(config.Mempool,
-				proxyAppConnConMem,
-				state.LastBlockHeight,
-				mempoolv0.WithPreCheck(sm.TxPreCheck(state)),
-				mempoolv0.WithPostCheck(sm.TxPostCheck(state)))
-		case cfg.MempoolV1:
-			mempool = mempoolv1.NewTxMempool(logger,
-				config.Mempool,
-				proxyAppConnConMem,
-				state.LastBlockHeight,
-				mempoolv1.WithPreCheck(sm.TxPreCheck(state)),
-				mempoolv1.WithPostCheck(sm.TxPostCheck(state)),
-			)
-		case cfg.MempoolV2:
-			mempool = mempoolv2.NewTxPool(
-				logger,
-				config.Mempool,
-				proxyAppConnConMem,
-				state.LastBlockHeight,
-				mempoolv2.WithPreCheck(sm.TxPreCheck(state)),
-				mempoolv2.WithPostCheck(sm.TxPostCheck(state)),
-			)
-		}
+		mempool := cat.NewTxPool(
+			logger,
+			config.Mempool,
+			proxyAppConnConMem,
+			state.LastBlockHeight,
+			cat.WithPreCheck(sm.TxPreCheck(state)),
+			cat.WithPostCheck(sm.TxPostCheck(state)),
+		)
+		var err error
+		catReactors[i], err = cat.NewReactor(mempool, &cat.ReactorOptions{
+			ListenOnly:     !config.Mempool.Broadcast,
+			MaxTxSize:      config.Mempool.MaxTxBytes,
+			MaxGossipDelay: config.Mempool.MaxGossipDelay,
+		})
+		require.NoError(t, err)
 
 		if thisConfig.Consensus.WaitForTxs() {
 			mempool.EnableTxsAvailable()
@@ -112,7 +97,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 		// Make State
 		blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool)
-		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, evpool)
+		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, catReactors[i], evpool)
 		cs.SetLogger(cs.Logger)
 		// set private validator
 		pv := privVals[i]
@@ -154,6 +139,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 	// make connected switches and start all reactors
 	p2p.MakeConnectedSwitches(config.P2P, nValidators, func(i int, s *p2p.Switch) *p2p.Switch {
 		s.AddReactor("CONSENSUS", reactors[i])
+		s.AddReactor("MEMPOOL", catReactors[i])
 		s.SetLogger(reactors[i].conS.Logger.With("module", "p2p"))
 		return s
 	}, p2p.Connect2Switches)
@@ -230,9 +216,15 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		}
 		proposerAddr := lazyProposer.privValidatorPubKey.Address()
 
-		block, blockParts := lazyProposer.blockExec.CreateProposalBlock(
+		block := lazyProposer.blockExec.CreateProposalBlock(
 			lazyProposer.Height, lazyProposer.state, commit, proposerAddr,
 		)
+		blockHash := block.Hash()
+		blockParts := block.MakePartSet(types.BlockPartSizeBytes)
+
+		keys, err := lazyProposer.txFetcher.FetchKeysFromTxs(context.Background(), block.Txs.ToSliceOfBytes())
+		require.NoError(t, err)
+		block.Txs = types.ToTxs(keys)
 
 		// Flush the WAL. Otherwise, we may not recompute the same proposal to sign,
 		// and the privValidator will refuse to sign anything.
@@ -241,14 +233,15 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		}
 
 		// Make proposal
-		propBlockID := types.BlockID{Hash: block.Hash(), PartSetHeader: blockParts.Header()}
+		propBlockID := types.BlockID{Hash: blockHash, PartSetHeader: blockParts.Header()}
 		proposal := types.NewProposal(height, round, lazyProposer.TwoThirdPrevoteRound, propBlockID)
 		p := proposal.ToProto()
 		if err := lazyProposer.privValidator.SignProposal(lazyProposer.state.ChainID, p); err == nil {
 			proposal.Signature = p.Signature
 
 			// send proposal and block parts on internal msg queue
 			lazyProposer.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
+			lazyProposer.sendInternalMessage(msgInfo{&CompactBlockMessage{block, proposal.Round}, ""})
 			for i := 0; i < int(blockParts.Total()); i++ {
 				part := blockParts.GetPart(i)
 				lazyProposer.sendInternalMessage(msgInfo{&BlockPartMessage{lazyProposer.Height, lazyProposer.Round, part}, ""})
@@ -307,7 +300,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		}
 	case <-time.After(20 * time.Second):
 		for i, reactor := range reactors {
-			t.Logf("Consensus Reactor %d\n%v", i, reactor)
+			t.Logf("Consensus Reactor %d\n%v", i, reactor.conS.GetRoundState())
 		}
 		t.Fatalf("Timed out waiting for validators to commit evidence")
 	}
@@ -480,7 +473,8 @@ func byzantineDecideProposalFunc(t *testing.T, height int64, round int32, cs *St
 	// Avoid sending on internalMsgQueue and running consensus state.
 
 	// Create a new proposal block from state/txs from the mempool.
-	block1, blockParts1 := cs.createProposalBlock()
+	block1 := cs.createProposalBlock()
+	blockParts1 := block1.MakePartSet(types.BlockPartSizeBytes)
 	polRound := cs.TwoThirdPrevoteRound
 	propBlockID := types.BlockID{Hash: block1.Hash(), PartSetHeader: blockParts1.Header()}
 	proposal1 := types.NewProposal(height, round, polRound, propBlockID)
@@ -495,7 +489,8 @@ func byzantineDecideProposalFunc(t *testing.T, height int64, round int32, cs *St
 	deliverTxsRange(cs, 0, 1)
 
 	// Create a new proposal block from state/txs from the mempool.
-	block2, blockParts2 := cs.createProposalBlock()
+	block2 := cs.createProposalBlock()
+	blockParts2 := block2.MakePartSet(types.BlockPartSizeBytes)
 	polRound = cs.TwoThirdPrevoteRound
 	propBlockID = types.BlockID{Hash: block2.Hash(), PartSetHeader: blockParts2.Header()}
 	proposal2 := types.NewProposal(height, round, polRound, propBlockID)

consensus/common_test.go

@@ -207,7 +207,9 @@ func decideProposal(
 	round int32,
 ) (proposal *types.Proposal, block *types.Block) {
 	cs1.mtx.Lock()
-	block, blockParts := cs1.createProposalBlock()
+	block = cs1.createProposalBlock()
+	blockParts := block.MakePartSet(types.BlockPartSizeBytes)
+
 	validRound := cs1.TwoThirdPrevoteRound
 	chainID := cs1.state.ChainID
 	cs1.mtx.Unlock()
@@ -447,7 +449,7 @@ func newStateWithConfigAndBlockStore(
 	}
 
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool)
-	cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, evpool)
+	cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, nil, evpool)
 	cs.SetLogger(log.TestingLogger().With("module", "consensus"))
 	cs.SetPrivValidator(pv)
 

consensus/metrics.go

@@ -1,12 +1,16 @@
 package consensus
 
 import (
+	"encoding/json"
+	"fmt"
+	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/go-kit/kit/metrics"
 	"github.com/go-kit/kit/metrics/discard"
 	cstypes "github.com/tendermint/tendermint/consensus/types"
+	"github.com/tendermint/tendermint/libs/os"
 
 	prometheus "github.com/go-kit/kit/metrics/prometheus"
 	stdprometheus "github.com/prometheus/client_golang/prometheus"
@@ -299,3 +303,49 @@ func (m *Metrics) MarkStep(s cstypes.RoundStepType) {
 	}
 	m.stepStart = time.Now()
 }
+
+type JSONMetrics struct {
+	dir                  string
+	interval             int
+	StartTime            time.Time
+	EndTime              time.Time
+	Blocks               uint64
+	Rounds               uint64
+	SentConsensusBytes   uint64
+	SentCompactBlocks    uint64
+	SentCompactBytes     uint64
+	CompactBlockFailures uint64
+	SentBlockParts       uint64
+	SentBlockPartsBytes  uint64
+}
+
+func NewJSONMetrics(dir string) *JSONMetrics {
+	return &JSONMetrics{
+		dir:       dir,
+		StartTime: time.Now().UTC(),
+	}
+}
+
+func (m *JSONMetrics) Save() {
+	m.EndTime = time.Now().UTC()
+	content, err := json.MarshalIndent(m, "", "  ")
+	if err != nil {
+		panic(err)
+	}
+	path := filepath.Join(m.dir, fmt.Sprintf("metrics_%d.json", m.interval))
+	os.MustWriteFile(path, content, 0644)
+	m.StartTime = m.EndTime
+	m.interval++
+	m.reset()
+}
+
+func (m *JSONMetrics) reset() {
+	m.Blocks = 0
+	m.Rounds = 0
+	m.SentConsensusBytes = 0
+	m.SentBlockParts = 0
+	m.SentBlockPartsBytes = 0
+	m.SentCompactBlocks = 0
+	m.SentCompactBytes = 0
+	m.CompactBlockFailures = 0
+}

consensus/msgs.go

@@ -60,6 +60,24 @@ func MsgToProto(msg Message) (*cmtcons.Message, error) {
 		}
 		return m.Wrap().(*cmtcons.Message), nil
 
+	case *CompactBlockMessage:
+		block, err := msg.Block.ToProto()
+		if err != nil {
+			return nil, fmt.Errorf("msg to proto error: %w", err)
+		}
+		m := &cmtcons.CompactBlock{
+			Block: block,
+			Round: msg.Round,
+		}
+		return m.Wrap().(*cmtcons.Message), nil
+
+	case *HasBlockMessage:
+		m := &cmtcons.HasCompactBlock{
+			Height: msg.Height,
+			Round:  msg.Round,
+		}
+		return m.Wrap().(*cmtcons.Message), nil
+
 	case *BlockPartMessage:
 		parts, err := msg.Part.ToProto()
 		if err != nil {
@@ -188,6 +206,22 @@ func MsgFromProto(p *cmtcons.Message) (Message, error) {
 			Round:  msg.Round,
 			Part:   parts,
 		}
+	case *cmtcons.CompactBlock:
+		block, err := types.BlockFromProto(msg.Block)
+		if err != nil {
+			return nil, fmt.Errorf("compactBlock msg to proto error: %w", err)
+		}
+		pb = &CompactBlockMessage{
+			Block: block,
+			Round: msg.Round,
+		}
+
+	case *cmtcons.HasCompactBlock:
+		pb = &HasBlockMessage{
+			Height: msg.Height,
+			Round:  msg.Round,
+		}
+
 	case *cmtcons.Vote:
 		vote, err := types.VoteFromProto(msg.Vote)
 		if err != nil {