revert some modifications to the consensus reactor logic

celestiaorg · cmwaters · Jan 20, 2024 · Jan 24, 2024 · Jan 24, 2024 · Jan 25, 2024
commit 77db5255c8669239a8c81e6ce92db99340b04f01
@@ -161,15 +161,15 @@
 		{
 			ID: DataChannel, // maybe split between gossiping current block and catchup stuff
 			// once we gossip the whole block there's nothing left to send until next height or round
-			Priority:            12,
-			SendQueueCapacity:   20,
+			Priority:            10,
+			SendQueueCapacity:   10,
 			RecvBufferCapacity:  50 * 4096,
 			RecvMessageCapacity: maxMsgSize,
 			MessageType:         &cmtcons.Message{},
 		},
 		{
 			ID:                  VoteChannel,
-			Priority:            8,
+			Priority:            7,
 			SendQueueCapacity:   100,
 			RecvBufferCapacity:  100 * 100,
 			RecvMessageCapacity: maxMsgSize,
@@ -596,7 +596,7 @@
 }

 // Broadcasts HasVoteMessage to peers that care.
 func (conR *Reactor) broadcastHasVoteMessage(vote *types.Vote) {
 	msg := &cmtcons.HasVote{
 		Height: vote.Height,
 		Round:  vote.Round,
@@ -702,7 +702,7 @@

 		switch {
 		case prs.Height == rs.Height && !prs.Proposal && rs.Proposal != nil: // same height no proposal
 			if p2p.TrySendEnvelopeShim(peer, p2p.Envelope{
 				ChannelID: DataChannel,
 				Message:   &cmtcons.Proposal{Proposal: *rs.Proposal.ToProto()},
 			}, logger) {
@@ -966,7 +966,7 @@
 ) bool {
 
 	// If there are lastCommits to send...
-	if prs.Step == cstypes.RoundStepNewHeight && rs.Step < cstypes.RoundStepPrevote {
+	if prs.Step == cstypes.RoundStepNewHeight {
 		if conR.pickSendVoteAndTrace(rs.LastCommit, rs, ps) {
 			logger.Debug("Picked rs.LastCommit to send")
 			return true
@@ -1365,7 +1365,8 @@
 // Returns the vote if vote was sent. Otherwise, returns nil.
 func (ps *PeerState) PickSendVote(votes types.VoteSetReader) *types.Vote {
 	if vote, ok := ps.PickVoteToSend(votes); ok {
-		if p2p.TrySendEnvelopeShim(ps.peer, p2p.Envelope{ //nolint: staticcheck
+		ps.logger.Debug("Sending vote message", "ps", ps, "vote", vote)
+		if p2p.SendEnvelopeShim(ps.peer, p2p.Envelope{ //nolint: staticcheck
 			ChannelID: VoteChannel,
 			Message: &cmtcons.Vote{
 				Vote: vote.ToProto(),

diff --git a/mempool/cat/reactor_test.go b/mempool/cat/reactor_test.go
 func TestReactorSendsTxAfterReceivingWantTx(t *testing.T) {
 	reactor, pool := setupReactor(t)
 	require.NoError(t, reactor.Start())
 	defer reactor.Stop()

 	tx := newDefaultTx("hello")
 	key := tx.Key()
diff --git a/mempool/cat/reactor.go b/mempool/cat/reactor.go
 		default:
 		}

 		// if there are no current transctions to broadcast, check if there are any new ones to add to the queue
 		if memR.mempool.priorityBroadcastQueue.queue.Len() == 0 {
 			select {
 			case <-memR.Quit():
 					break
 				}

 				if p2p.TrySendEnvelopeShim(peer, p2p.Envelope{
 					ChannelID: mempool.MempoolChannel,
 					Message:   &protomem.Txs{Txs: [][]byte{nextTx}},
 				}, memR.Logger) {
diff --git a/consensus/state.go b/consensus/state.go

 	// internal state
 	mtx cmtsync.RWMutex
 	cstypes.RoundState
 	state sm.State // State until height-1.
 	// privValidator pubkey, memoized for the duration of one block
 	// to avoid extra requests to HSM
 	}

 	// we start this as a goroutine so as to not block the reactor
 	// from recieving other messages or timeouts while the mempool
 	// attempts to fetch any missing transactions. If the block is
 	// correctly reconstructed, the handleCompleteProposal function
 	// will be called to process the block and move to the next step.
diff --git a/mempool/cat/pool.go b/mempool/cat/pool.go
 // allEntriesSorted returns a slice of all the transactions currently in the
 // mempool, sorted in nonincreasing order by priority with ties broken by
 // increasing order of arrival time.
 func (txmp *TxPool) allEntriesSorted() []*wrappedTx {
 	txs := txmp.store.getAllTxs()
 	sort.Slice(txs, PriorityFn(txs))
 	return txs
diff --git a/consensus/common_test.go b/consensus/common_test.go
 	case <-time.After(timeout):
 		break
 	case <-ch:
 		panic(fmt.Errorf(errorMessage))
 	}
 }

diff --git a/consensus/types/round_state.go b/consensus/types/round_state.go

 	// Last known round with POL for non-nil valid block.
 	TwoThirdPrevoteRound int32        `json:"valid_round"`
 	TwoThirdPrevoteBlock *types.Block `json:"valid_block"` // Last known block of POL mentioned above.

 	// Last known block parts of POL mentioned above.
 	TwoThirdPrevoteBlockParts *types.PartSet      `json:"valid_block_parts"`
	func TestReactorSendsTxAfterReceivingWantTx(t *testing.T) {
	reactor, pool := setupReactor(t)
	require.NoError(t, reactor.Start())
	defer reactor.Stop()
Check failure on line 92 in mempool/cat/reactor_test.go GitHub Actions / golangci-lint Error return value of `reactor.Stop` is not checked (errcheck)

	tx := newDefaultTx("hello")
	key := tx.Key()
	default:
	}

	// if there are no current transctions to broadcast, check if there are any new ones to add to the queue
Check failure on line 146 in mempool/cat/reactor.go GitHub Actions / golangci-lint `transctions` is a misspelling of `transitions` (misspell)
	if memR.mempool.priorityBroadcastQueue.queue.Len() == 0 {
	select {
	case <-memR.Quit():
	break
	}

	if p2p.TrySendEnvelopeShim(peer, p2p.Envelope{
Check failure on line 203 in mempool/cat/reactor.go GitHub Actions / golangci-lint `SA1019: p2p.TrySendEnvelopeShim is deprecated: Will be removed in v0.37. (staticcheck)`
	ChannelID: mempool.MempoolChannel,
	Message: &protomem.Txs{Txs: [][]byte{nextTx}},
	}, memR.Logger) {

	// internal state
	mtx cmtsync.RWMutex
	cstypes.RoundState
Check failure on line 128 in consensus/state.go GitHub Actions / golangci-lint `structtag: struct field TwoThirdPrevoteBlock repeats json tag "valid_block" also at round_state.go:84 (govet)`
	state sm.State // State until height-1.
	// privValidator pubkey, memoized for the duration of one block
	// to avoid extra requests to HSM
	}

	// we start this as a goroutine so as to not block the reactor
	// from recieving other messages or timeouts while the mempool
Check failure on line 2063 in consensus/state.go GitHub Actions / golangci-lint `recieving` is a misspelling of `receiving` (misspell)
	// attempts to fetch any missing transactions. If the block is
	// correctly reconstructed, the handleCompleteProposal function
	// will be called to process the block and move to the next step.
	// allEntriesSorted returns a slice of all the transactions currently in the
	// mempool, sorted in nonincreasing order by priority with ties broken by
	// increasing order of arrival time.
	func (txmp TxPool) allEntriesSorted() []wrappedTx {
Check failure on line 407 in mempool/cat/pool.go GitHub Actions / golangci-lint func `(*TxPool).allEntriesSorted` is unused (unused)
	txs := txmp.store.getAllTxs()
	sort.Slice(txs, PriorityFn(txs))
	return txs
	case <-time.After(timeout):
	break
	case <-ch:
	panic(fmt.Errorf(errorMessage))
Check failure on line 502 in consensus/common_test.go GitHub Actions / golangci-lint `printf: non-constant format string in call to fmt.Errorf (govet)`
	}
	}

	// Last known round with POL for non-nil valid block.
	TwoThirdPrevoteRound int32 `json:"valid_round"`
	TwoThirdPrevoteBlock *types.Block `json:"valid_block"` // Last known block of POL mentioned above.
Check failure on line 92 in consensus/types/round_state.go GitHub Actions / golangci-lint `structtag: struct field TwoThirdPrevoteBlock repeats json tag "valid_block" also at round_state.go:84 (govet)`

	// Last known block parts of POL mentioned above.
	TwoThirdPrevoteBlockParts *types.PartSet `json:"valid_block_parts"`