strongly typed clist

sei-tendermint/internal/consensus/mempool_test.go

@@ -147,10 +147,9 @@ func checkTxsRange(ctx context.Context, t *testing.T, cs *State, start, end int)
 	for i := start; i < end; i++ {
 		txBytes := make([]byte, 8)
 		binary.BigEndian.PutUint64(txBytes, uint64(i))
-		var rCode uint32
-		err := cs.txMempool.CheckTx(ctx, txBytes, func(r *abci.ResponseCheckTx) { rCode = r.Code }, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
 		require.NoError(t, err, "error after checkTx")
-		require.Equal(t, code.CodeTypeOK, rCode, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
+		require.Equal(t, code.CodeTypeOK, res.Code, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
 	}
 }
 
@@ -182,10 +181,9 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
 	for i := range int(numTxs) {
 		txBytes := make([]byte, 8)
 		binary.BigEndian.PutUint64(txBytes, uint64(i))
-		var rCode uint32
-		err := cs.txMempool.CheckTx(ctx, txBytes, func(r *abci.ResponseCheckTx) { rCode = r.Code }, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
 		require.NoError(t, err, "error after checkTx")
-		require.Equal(t, code.CodeTypeOK, rCode, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
+		require.Equal(t, code.CodeTypeOK, res.Code, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
 	}
 
 	startTestRound(ctx, cs, cs.roundState.Height(), cs.roundState.Round())
@@ -234,17 +232,16 @@ func TestMempoolRmBadTx(t *testing.T) {
 		// Try to send the tx through the mempool.
 		// CheckTx should not err, but the app should return a bad abci code
 		// and the tx should get removed from the pool
-		err := cs.txMempool.CheckTx(ctx, txBytes, func(r *abci.ResponseCheckTx) {
-			if r.Code != code.CodeTypeBadNonce {
-				t.Errorf("expected checktx to return bad nonce, got %v", r)
-				return
-			}
-			checkTxRespCh <- struct{}{}
-		}, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
 		if err != nil {
 			t.Errorf("error after CheckTx: %v", err)
 			return
 		}
+		if res.Code != code.CodeTypeBadNonce {
+			t.Errorf("expected checktx to return bad nonce, got %v", res)
+			return
+		}
+		checkTxRespCh <- struct{}{}
 
 		// check for the tx
 		for {

sei-tendermint/internal/consensus/reactor_test.go

@@ -151,7 +151,7 @@ func finalizeTx(
 	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
 		for i, sub := range blocksSubs {
 			s.Spawn(func() error {
-				if err := states[i].txMempool.CheckTx(ctx, tx, nil, mempool.TxInfo{}); err != nil {
+				if _, err := states[i].txMempool.CheckTx(ctx, tx, mempool.TxInfo{}); err != nil {
 					return fmt.Errorf("CheckTx(): %w", err)
 				}
 				for {
@@ -363,15 +363,12 @@ func TestReactorCreatesBlockWhenEmptyBlocksFalse(t *testing.T) {
 	}
 
 	// send a tx
-	require.NoError(
-		t,
-		states[1].txMempool.CheckTx(
-			ctx,
-			[]byte{1, 2, 3},
-			nil,
-			mempool.TxInfo{},
-		),
+	_, err := states[1].txMempool.CheckTx(
+		ctx,
+		[]byte{1, 2, 3},
+		mempool.TxInfo{},
 	)
+	require.NoError(t, err)
 
 	var wg sync.WaitGroup
 	for _, sub := range rts.subs {

sei-tendermint/internal/consensus/replay_test.go

@@ -152,7 +152,7 @@ func sendTxs(ctx context.Context, cs *State) error {
 			return nil
 		}
 		tx := []byte{byte(i)}
-		if err := cs.txMempool.CheckTx(ctx, tx, nil, mempool.TxInfo{}); err != nil {
+		if _, err := cs.txMempool.CheckTx(ctx, tx, mempool.TxInfo{}); err != nil {
 			return fmt.Errorf("cs.mempool.CheckTx(): %w", err)
 		}
 	}
@@ -347,7 +347,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	valPubKey1ABCI := crypto.PubKeyToProto(newValidatorPubKey1)
 	newValidatorTx1 := kvstore.MakeValSetChangeTx(valPubKey1ABCI, testMinPower)
-	err = css[0].txMempool.CheckTx(ctx, newValidatorTx1, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx1, mempool.TxInfo{})
 	assert.NoError(t, err)
 	propBlock, err := css[0].createProposalBlock(ctx) // changeProposer(t, cs1, vs2)
 	require.NoError(t, err)
@@ -383,7 +383,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	updatePubKey1ABCI := crypto.PubKeyToProto(updateValidatorPubKey1)
 	updateValidatorTx1 := kvstore.MakeValSetChangeTx(updatePubKey1ABCI, 25)
-	err = css[0].txMempool.CheckTx(ctx, updateValidatorTx1, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, updateValidatorTx1, mempool.TxInfo{})
 	assert.NoError(t, err)
 	propBlock, err = css[0].createProposalBlock(ctx) // changeProposer(t, cs1, vs2)
 	require.NoError(t, err)
@@ -418,14 +418,14 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	newVal2ABCI := crypto.PubKeyToProto(newValidatorPubKey2)
 	newValidatorTx2 := kvstore.MakeValSetChangeTx(newVal2ABCI, testMinPower)
-	err = css[0].txMempool.CheckTx(ctx, newValidatorTx2, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx2, mempool.TxInfo{})
 	assert.NoError(t, err)
 	pv, _ = css[nVals+2].privValidator.Get()
 	newValidatorPubKey3, err := pv.GetPubKey(ctx)
 	require.NoError(t, err)
 	newVal3ABCI := crypto.PubKeyToProto(newValidatorPubKey3)
 	newValidatorTx3 := kvstore.MakeValSetChangeTx(newVal3ABCI, testMinPower)
-	err = css[0].txMempool.CheckTx(ctx, newValidatorTx3, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx3, mempool.TxInfo{})
 	assert.NoError(t, err)
 	propBlock, err = css[0].createProposalBlock(ctx) // changeProposer(t, cs1, vs2)
 	require.NoError(t, err)
@@ -471,7 +471,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	ensureNewProposal(t, proposalCh, height, round)
 
 	removeValidatorTx2 := kvstore.MakeValSetChangeTx(newVal2ABCI, 0)
-	err = css[0].txMempool.CheckTx(ctx, removeValidatorTx2, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx2, mempool.TxInfo{})
 	assert.NoError(t, err)
 
 	rs = css[0].GetRoundState()
@@ -515,7 +515,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	height++
 	incrementHeight(vss...)
 	removeValidatorTx3 := kvstore.MakeValSetChangeTx(newVal3ABCI, 0)
-	err = css[0].txMempool.CheckTx(ctx, removeValidatorTx3, nil, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx3, mempool.TxInfo{})
 	assert.NoError(t, err)
 	propBlock, err = css[0].createProposalBlock(ctx) // changeProposer(t, cs1, vs2)
 	require.NoError(t, err)

sei-tendermint/internal/consensus/state_test.go

@@ -2230,7 +2230,7 @@ func TestStartNextHeightCorrectlyAfterTimeout(t *testing.T) {
 
 	ensureNewBlockHeader(t, newBlockHeader, height, blockID.Hash)
 
-	err := cs1.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), nil, mempool.TxInfo{})
+	_, err := cs1.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), mempool.TxInfo{})
 	require.NoError(t, err, "failed to seed the mempool with a transaction")
 
 	ensureNewTimeout(t, timeoutProposeCh, height+1, round)
@@ -2650,7 +2650,7 @@ func TestTryCreateProposalBlock_PartsMismatch(t *testing.T) {
 	incrementRound(vss[1:]...)
 	startTestRound(ctx, cs, height, round)
 
-	err := cs.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), nil, mempool.TxInfo{})
+	_, err := cs.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), mempool.TxInfo{})
 	require.NoError(t, err, "failed to seed the mempool with a transaction")
 
 	proposal, block := decideProposal(ctx, t, cs, vss[1], height, round)

sei-tendermint/internal/evidence/doc.go

@@ -17,7 +17,7 @@ The `Receive` function takes a list of evidence and does the following:
 3. Stores the evidence to a db and a concurrent list
 
 The gossiping of evidence is initiated when a peer is added which starts a go routine to broadcast currently
-uncommitted evidence at intervals of 60 seconds (set by the by broadcastEvidenceIntervalS).
+uncommitted evidence at intervals of 60 seconds (set by `broadcastEvidenceInterval`).
 It uses a concurrent list to store the evidence and before sending verifies that each evidence is still valid in the
 sense that it has not exceeded the max evidence age and height (see types/params.go#EvidenceParams).
 

sei-tendermint/internal/evidence/pool.go

@@ -44,8 +44,8 @@ var logger = seilog.NewLogger("tendermint", "internal", "evidence")
 // Pool maintains a pool of valid evidence to be broadcasted and committed
 type Pool struct {
 	evidenceStore dbm.DB
-	evidenceList  *clist.CList // concurrent linked-list of evidence
-	evidenceSize  uint32       // amount of pending evidence
+	evidenceList  *clist.CList[types.Evidence] // concurrent linked-list of evidence
+	evidenceSize  uint32                       // amount of pending evidence
 
 	// needed to load headers and commits to verify evidence
 	blockStore BlockStore
@@ -78,7 +78,7 @@ func NewPool(evidenceDB dbm.DB, stateStore sm.Store, blockStore BlockStore, metr
 		blockStore:      blockStore,
 		stateDB:         stateStore,
 		evidenceStore:   evidenceDB,
-		evidenceList:    clist.New(),
+		evidenceList:    clist.New[types.Evidence](),
 		consensusBuffer: make([]duplicateVoteSet, 0),
 		Metrics:         metrics,
 		eventBus:        eventBus,
@@ -238,14 +238,13 @@ func (evpool *Pool) CheckEvidence(ctx context.Context, evList types.EvidenceList
 }
 
 // EvidenceFront goes to the first evidence in the clist
-func (evpool *Pool) EvidenceFront() *clist.CElement {
+func (evpool *Pool) EvidenceFront() *clist.CElement[types.Evidence] {
 	return evpool.evidenceList.Front()
 }
 
-// EvidenceWaitChan is a channel that closes once the first evidence in the list
-// is there. i.e Front is not nil.
-func (evpool *Pool) EvidenceWaitChan() <-chan struct{} {
-	return evpool.evidenceList.WaitChan()
+// WaitEvidenceFront waits until the first evidence in the list is available.
+func (evpool *Pool) WaitEvidenceFront(ctx context.Context) (*clist.CElement[types.Evidence], error) {
+	return evpool.evidenceList.WaitFront(ctx)
 }
 
 // Size returns the number of evidence in the pool.
@@ -532,7 +531,7 @@ func (evpool *Pool) removeEvidenceFromList(
 
 	for e := evpool.evidenceList.Front(); e != nil; e = e.Next() {
 		// Remove from clist
-		ev := e.Value.(types.Evidence)
+		ev := e.Value()
 		if _, ok := blockEvidenceMap[evMapKey(ev)]; ok {
 			evpool.evidenceList.Remove(e)
 			e.DetachPrev()

sei-tendermint/internal/evidence/pool_test.go

@@ -86,7 +86,8 @@ func TestEvidencePoolBasic(t *testing.T) {
 	// good evidence
 	evAdded := make(chan struct{})
 	go func() {
-		<-pool.EvidenceWaitChan()
+		_, err := pool.WaitEvidenceFront(ctx)
+		require.NoError(t, err)
 		close(evAdded)
 	}()
 
@@ -101,7 +102,7 @@ func TestEvidencePoolBasic(t *testing.T) {
 	}
 
 	next := pool.EvidenceFront()
-	require.Equal(t, ev, next.Value.(types.Evidence))
+	require.Equal(t, ev, next.Value())
 
 	const evidenceBytes int64 = 372
 	evs, size = pool.PendingEvidence(evidenceBytes)
@@ -505,7 +506,7 @@ func TestRecoverPendingEvidence(t *testing.T) {
 	require.Equal(t, 1, len(evList))
 
 	next := newPool.EvidenceFront()
-	require.Equal(t, goodEvidence, next.Value.(types.Evidence))
+	require.Equal(t, goodEvidence, next.Value())
 }
 
 func initializeStateFromValidatorSet(t *testing.T, valSet *types.ValidatorSet, height int64) sm.Store {

sei-tendermint/internal/evidence/reactor.go

@@ -2,6 +2,7 @@ package evidence
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"runtime/debug"
 	"sync"
@@ -10,6 +11,7 @@ import (
 	clist "github.com/sei-protocol/sei-chain/sei-tendermint/internal/libs/clist"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/service"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	pb "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/types"
 )
@@ -25,7 +27,7 @@ const (
 	// goes back to the start of the list and begins sending the evidence again.
 	// Most evidence should be committed in the very next block that is why we wait
 	// just over the block production rate before sending evidence again.
-	broadcastEvidenceIntervalS = 10
+	broadcastEvidenceInterval = 60 * time.Second
 )
 
 // GetChannelDescriptor produces an instance of a descriptor for this
@@ -213,7 +215,6 @@ func (r *Reactor) processPeerUpdates(ctx context.Context) error {
 //
 // REF: https://github.com/tendermint/tendermint/issues/4727
 func (r *Reactor) broadcastEvidenceLoop(ctx context.Context, peerID types.NodeID, evidenceCh *p2p.Channel[*pb.Evidence]) {
-	var next *clist.CElement
 
 	defer func() {
 		r.mtx.Lock()
@@ -229,50 +230,44 @@ func (r *Reactor) broadcastEvidenceLoop(ctx context.Context, peerID types.NodeID
 		}
 	}()
 
-	timer := time.NewTimer(0)
-	defer timer.Stop()
-
 	for {
 		// This happens because the CElement we were looking at got garbage
-		// collected (removed). That is, .NextWaitChan() returned nil. So we can go
+		// collected (removed). That is, NextWait() returned nil. So we can go
 		// ahead and start from the beginning.
-		if next == nil {
-			select {
-			case <-r.evpool.EvidenceWaitChan(): // wait until next evidence is available
-				if next = r.evpool.EvidenceFront(); next == nil {
-					continue
-				}
-
-			case <-ctx.Done():
-				return
-			}
-		}
-
-		ev := next.Value.(types.Evidence)
-		evProto, err := types.EvidenceToProto(ev)
+		next, err := r.evpool.WaitEvidenceFront(ctx)
 		if err != nil {
-			panic(fmt.Errorf("failed to convert evidence: %w", err))
+			return
 		}
 
-		// Send the evidence to the corresponding peer. Note, the peer may be behind
-		// and thus would not be able to process the evidence correctly. Also, the
-		// peer may receive this piece of evidence multiple times if it added and
-		// removed frequently from the broadcasting peer.
-
-		evidenceCh.Send(evProto, peerID)
-		logger.Debug("gossiped evidence to peer", "evidence", ev, "peer", peerID)
-
-		select {
-		case <-timer.C:
-			// start from the beginning after broadcastEvidenceIntervalS seconds
-			timer.Reset(time.Second * broadcastEvidenceIntervalS)
-			next = nil
+		err = utils.WithTimeout(ctx, broadcastEvidenceInterval, func(ctx context.Context) error {
+			for {
+				ev := next.Value()
+				evProto, err := types.EvidenceToProto(ev)
+				if err != nil {
+					// This should never happen, but for some historical reasons there is a recover() deferred
+					// in this function. Current behavior is that evidence will stop being broadcasted to a peer
+					// if this happens, but the node with not crash.
+					// TODO(gprusak): reevaluate if this broadcastEvidenceLoop may panic even if there is no bug.
+					//   If it cannot, remove recover().
+					panic(fmt.Errorf("failed to convert evidence: %w", err))
+				}
 
-		case <-next.NextWaitChan():
-			next = next.Next()
-			timer.Stop()
+				// Send the evidence to the corresponding peer. Note, the peer may be behind
+				// and thus would not be able to process the evidence correctly. Also, the
+				// peer may receive this piece of evidence multiple times if it added and
+				// removed frequently from the broadcasting peer.
+				evidenceCh.Send(evProto, peerID)
+				logger.Debug("gossiped evidence to peer", "evidence", ev, "peer", peerID)
 
-		case <-ctx.Done():
+				next, err = next.NextWait(ctx)
+				if err != nil {
+					return err
+				}
+			}
+		})
+		// In case the observed element has been removed or we waited too long for the next element,
+		// we start sending from the beginning.
+		if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, clist.ErrRemoved) {
 			return
 		}
 	}

sei-tendermint/internal/libs/clist/bench_test.go

@@ -3,7 +3,7 @@ package clist
 import "testing"
 
 func BenchmarkDetaching(b *testing.B) {
-	lst := New()
+	lst := New[int]()
 	for i := 0; i < b.N+1; i++ {
 		lst.PushBack(i)
 	}
@@ -22,7 +22,7 @@ func BenchmarkDetaching(b *testing.B) {
 
 // This is used to benchmark the time of RMutex.
 func BenchmarkRemoved(b *testing.B) {
-	lst := New()
+	lst := New[int]()
 	for i := 0; i < b.N+1; i++ {
 		lst.PushBack(i)
 	}
@@ -38,7 +38,7 @@ func BenchmarkRemoved(b *testing.B) {
 }
 
 func BenchmarkPushBack(b *testing.B) {
-	lst := New()
+	lst := New[int]()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		lst.PushBack(i)