Platformvm network refactor

scripts/mocks.mockgen.txt

@@ -33,6 +33,7 @@ github.com/ava-labs/avalanchego/vms/components/avax=TransferableOut=vms/componen
 github.com/ava-labs/avalanchego/vms/components/verify=Verifiable=vms/components/verify/mock_verifiable.go
 github.com/ava-labs/avalanchego/vms/platformvm/blocks/executor=Manager=vms/platformvm/blocks/executor/mock_manager.go
 github.com/ava-labs/avalanchego/vms/platformvm/blocks=Block=vms/platformvm/blocks/mock_block.go
+github.com/ava-labs/avalanchego/vms/platformvm/blocks/builder=Builder=vms/platformvm/blocks/builder/mock_builder.go
 github.com/ava-labs/avalanchego/vms/platformvm/fx=Fx,Owner=vms/platformvm/fx/mock_fx.go
 github.com/ava-labs/avalanchego/vms/platformvm/state=Chain=vms/platformvm/state/mock_chain.go
 github.com/ava-labs/avalanchego/vms/platformvm/state=Diff=vms/platformvm/state/mock_diff.go

vms/avm/network/network.go

@@ -86,7 +86,7 @@ func (n *network) AppGossip(ctx context.Context, nodeID ids.NodeID, msgBytes []b
 		return nil
 	}
 
-	msg, ok := msgIntf.(*message.Tx)
+	msg, ok := msgIntf.(*message.TxGossip)
 	if !ok {
 		n.ctx.Log.Debug("dropping unexpected message",
 			zap.Stringer("nodeID", nodeID),
@@ -125,7 +125,7 @@ func (n *network) IssueTx(ctx context.Context, tx *txs.Tx) error {
 	}
 
 	txBytes := tx.Bytes()
-	msg := &message.Tx{
+	msg := &message.TxGossip{
 		Tx: txBytes,
 	}
 	msgBytes, err := message.Build(msg)

vms/avm/network/network_test.go

@@ -74,7 +74,7 @@ func TestNetworkAppGossip(t *testing.T) {
 			// Shouldn't attempt to issue or gossip the tx
 			name: "invalid tx bytes",
 			msgBytesFunc: func() []byte {
-				msg := message.Tx{
+				msg := message.TxGossip{
 					Tx: []byte{0x00},
 				}
 				msgBytes, err := message.Build(&msg)
@@ -94,7 +94,7 @@ func TestNetworkAppGossip(t *testing.T) {
 			// Issue returns nil because mempool has tx. We should gossip the tx.
 			name: "issuance succeeds",
 			msgBytesFunc: func() []byte {
-				msg := message.Tx{
+				msg := message.TxGossip{
 					Tx: testTx.Bytes(),
 				}
 				msgBytes, err := message.Build(&msg)
@@ -116,7 +116,7 @@ func TestNetworkAppGossip(t *testing.T) {
 			// Issue returns error because tx was dropped. We shouldn't gossip the tx.
 			name: "issuance fails",
 			msgBytesFunc: func() []byte {
-				msg := message.Tx{
+				msg := message.TxGossip{
 					Tx: testTx.Bytes(),
 				}
 				msgBytes, err := message.Build(&msg)

vms/components/message/codec.go

@@ -25,7 +25,7 @@ func init() {
 
 	errs := wrappers.Errs{}
 	errs.Add(
-		lc.RegisterType(&Tx{}),
+		lc.RegisterType(&TxGossip{}),
 		c.RegisterCodec(codecVersion, lc),
 	)
 	if errs.Errored() {

vms/components/message/handler.go

@@ -12,18 +12,18 @@ import (
 
 var _ Handler = NoopHandler{}
 
+// Handler handles incoming gossip messages
 type Handler interface {
-	HandleTx(nodeID ids.NodeID, requestID uint32, msg *Tx) error
+	HandleTxGossip(nodeID ids.NodeID, msg *TxGossip) error
 }
 
 type NoopHandler struct {
 	Log logging.Logger
 }
 
-func (h NoopHandler) HandleTx(nodeID ids.NodeID, requestID uint32, _ *Tx) error {
+func (h NoopHandler) HandleTxGossip(nodeID ids.NodeID, _ *TxGossip) error {
 	h.Log.Debug("dropping unexpected Tx message",
 		zap.Stringer("nodeID", nodeID),
-		zap.Uint32("requestID", requestID),
 	)
 	return nil
 }

vms/components/message/handler_test.go

@@ -16,16 +16,16 @@ type CounterHandler struct {
 	Tx int
 }
 
-func (h *CounterHandler) HandleTx(ids.NodeID, uint32, *Tx) error {
+func (h *CounterHandler) HandleTxGossip(ids.NodeID, *TxGossip) error {
 	h.Tx++
 	return nil
 }
 
-func TestHandleTx(t *testing.T) {
+func TestHandleTxGossip(t *testing.T) {
 	require := require.New(t)
 
 	handler := CounterHandler{}
-	msg := Tx{}
+	msg := TxGossip{}
 
 	err := msg.Handle(&handler, ids.EmptyNodeID, 0)
 	require.NoError(err)
@@ -37,6 +37,6 @@ func TestNoopHandler(t *testing.T) {
 		Log: logging.NoLog{},
 	}
 
-	err := handler.HandleTx(ids.EmptyNodeID, 0, nil)
+	err := handler.HandleTxGossip(ids.EmptyNodeID, nil)
 	require.NoError(t, err)
 }

vms/components/message/message.go

@@ -15,7 +15,7 @@ import (
 )
 
 var (
-	_ Message = (*Tx)(nil)
+	_ Message = (*TxGossip)(nil)
 
 	ErrUnexpectedCodecVersion = errors.New("unexpected codec version")
 	errUnknownMessageType     = errors.New("unknown message type")
@@ -54,7 +54,7 @@ func Parse(bytes []byte) (Message, error) {
 		// This message was encoded with proto.
 		switch m := protoMsg.GetMessage().(type) {
 		case *pb.Message_Tx:
-			msg = &Tx{
+			msg = &TxGossip{
 				Tx: m.Tx.Tx,
 			}
 		default:

vms/components/message/message_test.go

@@ -38,7 +38,7 @@ func TestParseProto(t *testing.T) {
 	parsedMsgIntf, err := Parse(msgBytes)
 	require.NoError(err)
 
-	parsedMsg, ok := parsedMsgIntf.(*Tx)
+	parsedMsg, ok := parsedMsgIntf.(*TxGossip)
 	require.True(ok)
 
 	require.Equal(txBytes, parsedMsg.Tx)

vms/components/message/tx.go → vms/components/message/tx_gossip.go

@@ -7,14 +7,14 @@ import (
 	"github.com/ava-labs/avalanchego/ids"
 )
 
-var _ Message = (*Tx)(nil)
+var _ Message = (*TxGossip)(nil)
 
-type Tx struct {
+type TxGossip struct {
 	message
 
 	Tx []byte `serialize:"true"`
 }
 
-func (msg *Tx) Handle(handler Handler, nodeID ids.NodeID, requestID uint32) error {
-	return handler.HandleTx(nodeID, requestID, msg)
+func (msg *TxGossip) Handle(handler Handler, nodeID ids.NodeID, _ uint32) error {
+	return handler.HandleTxGossip(nodeID, msg)
 }

vms/components/message/tx_test.go → vms/components/message/tx_gossip_test.go

@@ -12,11 +12,11 @@ import (
 	"github.com/ava-labs/avalanchego/utils/units"
 )
 
-func TestTx(t *testing.T) {
+func TestTxGossip(t *testing.T) {
 	require := require.New(t)
 
 	tx := utils.RandomBytes(256 * units.KiB)
-	builtMsg := Tx{
+	builtMsg := TxGossip{
 		Tx: tx,
 	}
 	builtMsgBytes, err := Build(&builtMsg)
@@ -27,7 +27,7 @@ func TestTx(t *testing.T) {
 	require.NoError(err)
 	require.Equal(builtMsgBytes, parsedMsgIntf.Bytes())
 
-	parsedMsg, ok := parsedMsgIntf.(*Tx)
+	parsedMsg, ok := parsedMsgIntf.(*TxGossip)
 	require.True(ok)
 
 	require.Equal(tx, parsedMsg.Tx)

vms/platformvm/blocks/builder/builder.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ava-labs/avalanchego/utils/timer/mockable"
 	"github.com/ava-labs/avalanchego/utils/units"
 	"github.com/ava-labs/avalanchego/vms/platformvm/blocks"
+	"github.com/ava-labs/avalanchego/vms/platformvm/network/sender"
 	"github.com/ava-labs/avalanchego/vms/platformvm/state"
 	"github.com/ava-labs/avalanchego/vms/platformvm/txs"
 	"github.com/ava-labs/avalanchego/vms/platformvm/txs/mempool"
@@ -42,7 +43,6 @@ var (
 type Builder interface {
 	mempool.Mempool
 	mempool.BlockTimer
-	Network
 
 	// set preferred block on top of which we'll build next
 	SetPreference(blockID ids.ID)
@@ -64,7 +64,8 @@ type Builder interface {
 // builder implements a simple builder to convert txs into valid blocks
 type builder struct {
 	mempool.Mempool
-	Network
+
+	networkClient sender.Sender
 
 	txBuilder         txbuilder.Builder
 	txExecutorBackend *txexecutor.Backend
@@ -83,30 +84,25 @@ type builder struct {
 }
 
 func New(
-	mempool mempool.Mempool,
 	txBuilder txbuilder.Builder,
 	txExecutorBackend *txexecutor.Backend,
 	blkManager blockexecutor.Manager,
 	toEngine chan<- common.Message,
-	appSender common.AppSender,
+	networkClient sender.Sender,
 ) Builder {
 	builder := &builder{
-		Mempool:           mempool,
+		Mempool:           txExecutorBackend.Mempool,
 		txBuilder:         txBuilder,
 		txExecutorBackend: txExecutorBackend,
 		blkManager:        blkManager,
 		toEngine:          toEngine,
+		networkClient:     networkClient,
 	}
 
 	builder.timer = timer.NewTimer(builder.setNextBuildBlockTime)
 
-	builder.Network = NewNetwork(
-		txExecutorBackend.Ctx,
-		builder,
-		appSender,
-	)
+	go builder.txExecutorBackend.Ctx.Log.RecoverAndPanic(builder.timer.Dispatch)
 
-	go txExecutorBackend.Ctx.Log.RecoverAndPanic(builder.timer.Dispatch)
 	return builder
 }
 
@@ -150,7 +146,7 @@ func (b *builder) AddUnverifiedTx(tx *txs.Tx) error {
 	if err := b.Mempool.Add(tx); err != nil {
 		return err
 	}
-	return b.GossipTx(tx)
+	return b.networkClient.GossipTx(tx)
 }
 
 // BuildBlock builds a block to be added to consensus.
@@ -166,7 +162,7 @@ func (b *builder) BuildBlock(context.Context) (snowman.Block, error) {
 	ctx := b.txExecutorBackend.Ctx
 	ctx.Log.Debug("starting to attempt to build a block")
 
-	statelessBlk, err := b.buildBlock()
+	statelessBlk, err := b.buildNextBlock()
 	if err != nil {
 		return nil, err
 	}
@@ -180,7 +176,7 @@ func (b *builder) BuildBlock(context.Context) (snowman.Block, error) {
 
 // Returns the block we want to build and issue.
 // Only modifies state to remove expired proposal txs.
-func (b *builder) buildBlock() (blocks.Block, error) {
+func (b *builder) buildNextBlock() (blocks.Block, error) {
 	// Get the block to build on top of and retrieve the new block's context.
 	preferred, err := b.Preferred()
 	if err != nil {
@@ -285,7 +281,7 @@ func (b *builder) setNextBuildBlockTime() {
 		return
 	}
 
-	if _, err := b.buildBlock(); err == nil {
+	if _, err := b.buildNextBlock(); err == nil {
 		// We can build a block now
 		b.notifyBlockReady()
 		return