more unit tests

gossip/gossip_test.go

@@ -4,19 +4,102 @@
 package gossip
 
 import (
+	"context"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/ava-labs/avalanchego/codec"
+	"github.com/ava-labs/avalanchego/codec/linearcodec"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p"
+	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/avalanchego/utils/set"
+	"github.com/ava-labs/avalanchego/utils/units"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
 )
 
 func TestGossiperShutdown(t *testing.T) {
 	config := Config{Frequency: time.Second}
-	puller := NewGossiper[testTx, *testTx](config, nil, nil, nil, 0)
+	gossiper := NewGossiper[testTx](config, nil, nil, nil, 0)
 	done := make(chan struct{})
 	wg := &sync.WaitGroup{}
 
 	wg.Add(1)
-	go puller.Gossip(done, wg)
+	go gossiper.Gossip(done, wg)
+
+	close(done)
+	wg.Wait()
+}
+
+func TestGossiperGossip(t *testing.T) {
+	require := require.New(t)
+	ctrl := gomock.NewController(t)
+
+	cc := codec.NewManager(2 * units.MiB)
+	lc := linearcodec.NewDefault()
+	require.NoError(lc.RegisterType(PullGossipRequest{}))
+	require.NoError(lc.RegisterType(PullGossipResponse{}))
+	require.NoError(cc.RegisterCodec(0, lc))
+
+	responseSender := common.NewMockSender(ctrl)
+	responseRouter := p2p.NewRouter(logging.NoLog{}, responseSender)
+	responseSet := testSet{
+		set: set.Set[*testTx]{},
+	}
+	tx := &testTx{hash: Hash{1, 2, 3, 4, 5}}
+	require.NoError(responseSet.Add(tx))
+	handler := NewHandler[*testTx](responseSet, cc, 0)
+	_, err := responseRouter.RegisterAppProtocol(0x0, handler)
+	require.NoError(err)
+
+	requestSender := common.NewMockSender(ctrl)
+	requestRouter := p2p.NewRouter(logging.NoLog{}, requestSender)
+	require.NoError(requestRouter.Connected(context.Background(), ids.EmptyNodeID, nil))
+
+	gossiped := make(chan struct{})
+	requestSender.EXPECT().SendAppRequest(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		Do(func(ctx context.Context, nodeIDs set.Set[ids.NodeID], requestID uint32, request []byte) {
+			go func() {
+				require.NoError(responseRouter.AppRequest(ctx, ids.EmptyNodeID, requestID, time.Time{}, request))
+			}()
+		}).AnyTimes()
+
+	responseSender.EXPECT().
+		SendAppResponse(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		Do(func(ctx context.Context, nodeID ids.NodeID, requestID uint32, appResponseBytes []byte) {
+			go func() {
+				require.NoError(requestRouter.AppResponse(ctx, nodeID, requestID, appResponseBytes))
+				close(gossiped)
+			}()
+		}).AnyTimes()
+
+	bloom, err := NewBloomFilter(1000, 0.01)
+	require.NoError(err)
+	requestSet := testSet{
+		set:   set.Set[*testTx]{},
+		bloom: bloom,
+	}
+	requestClient, err := requestRouter.RegisterAppProtocol(0x0, nil)
+	require.NoError(err)
+
+	config := Config{
+		Frequency: time.Second,
+		PollSize:  1,
+	}
+	gossiper := NewGossiper[testTx, *testTx](config, requestSet, requestClient, cc, 0)
+	done := make(chan struct{})
+	wg := &sync.WaitGroup{}
+
+	require.Len(requestSet.set, 0)
+	go gossiper.Gossip(done, wg)
+	<-gossiped
+
+	require.Len(requestSet.set, 1)
+	gotTx, _ := requestSet.set.Pop()
+	require.Equal(tx, gotTx)
 
 	close(done)
 	wg.Wait()

gossip/test_gossip.go

@@ -3,7 +3,14 @@
 
 package gossip
 
-var _ Gossipable = (*testTx)(nil)
+import (
+	"github.com/ava-labs/avalanchego/utils/set"
+)
+
+var (
+	_ Gossipable   = (*testTx)(nil)
+	_ Set[*testTx] = (*testSet)(nil)
+)
 
 type testTx struct {
 	hash Hash
@@ -14,11 +21,37 @@ func (t *testTx) GetHash() Hash {
 }
 
 func (t *testTx) Marshal() ([]byte, error) {
-	// TODO implement me
-	panic("implement me")
+	return t.hash[:], nil
 }
 
 func (t *testTx) Unmarshal(bytes []byte) error {
-	// TODO implement me
-	panic("implement me")
+	t.hash = Hash{}
+	copy(t.hash[:], bytes[:])
+	return nil
+}
+
+type testSet struct {
+	set   set.Set[*testTx]
+	bloom *BloomFilter
+}
+
+func (t testSet) Add(gossipable *testTx) error {
+	t.set.Add(gossipable)
+	return nil
+}
+
+func (t testSet) Get(filter func(gossipable *testTx) bool) []*testTx {
+	result := make([]*testTx, 0)
+	for tx := range t.set {
+		if !filter(tx) {
+			continue
+		}
+		result = append(result, tx)
+	}
+
+	return result
+}
+
+func (t testSet) GetFilter() *BloomFilter {
+	return t.bloom
 }

plugin/evm/message/codec.go

@@ -20,6 +20,7 @@ const (
 var (
 	Codec           codec.Manager
 	CrossChainCodec codec.Manager
+	SdkCodec        codec.Manager
 )
 
 func init() {
@@ -69,4 +70,6 @@ func init() {
 	if errs.Errored() {
 		panic(errs.Err)
 	}
+
+	SdkCodec = codec.NewManager(maxMessageSize)
 }

plugin/evm/vm.go

@@ -993,7 +993,7 @@ func (vm *VM) initBlockBuilding() error {
 	}
 	vm.atomicTxGossipClient = atomicTxGossipClient
 
-	vm.ethTxGossiper = gossip.NewGossiper[GossipEthTx, *GossipEthTx](
+	vm.ethTxGossiper = gossip.NewGossiper[GossipEthTx](
 		txGossipConfig,
 		ethTxPool,
 		vm.ethTxGossipClient,
@@ -1003,7 +1003,7 @@ func (vm *VM) initBlockBuilding() error {
 	vm.shutdownWg.Add(1)
 	go vm.ethTxGossiper.Gossip(vm.shutdownChan, &vm.shutdownWg)
 
-	vm.atomicTxGossiper = gossip.NewGossiper[GossipAtomicTx, *GossipAtomicTx](
+	vm.atomicTxGossiper = gossip.NewGossiper[GossipAtomicTx](
 		txGossipConfig,
 		vm.mempool,
 		vm.atomicTxGossipClient,