Comm Component for Simplex

message/outbound_msg_builder.go

@@ -181,6 +181,10 @@ type OutboundMsgBuilder interface {
 		chainID ids.ID,
 		msg []byte,
 	) (OutboundMessage, error)
+
+	SimplexMessage(
+		msg *p2p.Simplex,
+	) (OutboundMessage, error)
 }
 
 type outMsgBuilder struct {
@@ -725,3 +729,15 @@ func (b *outMsgBuilder) AppGossip(chainID ids.ID, msg []byte) (OutboundMessage,
 		false,
 	)
 }
+
+func (b *outMsgBuilder) SimplexMessage(msg *p2p.Simplex) (OutboundMessage, error) {
+	return b.builder.createOutbound(
+		&p2p.Message{
+			Message: &p2p.Message_Simplex{
+				Simplex: msg,
+			},
+		},
+		b.compressionType,
+		false,
+	)
+}

simplex/bls_test.go

@@ -13,8 +13,7 @@ import (
 )
 
 func TestBLSVerifier(t *testing.T) {
-	config, err := newEngineConfig()
-	require.NoError(t, err)
+	config := newEngineConfig(t, 1)
 	signer, verifier := NewBLSAuth(config)
 	otherNodeID := ids.GenerateTestNodeID()
 
@@ -81,7 +80,7 @@ func TestBLSVerifier(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err = verifier.Verify(msg, tt.sig, tt.nodeID)
+			err := verifier.Verify(msg, tt.sig, tt.nodeID)
 			require.ErrorIs(t, err, tt.expectErr)
 		})
 	}

simplex/comm.go

@@ -0,0 +1,143 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package simplex
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"slices"
+
+	"github.com/ava-labs/simplex"
+	"go.uber.org/zap"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/message"
+	"github.com/ava-labs/avalanchego/proto/pb/p2p"
+	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/snow/networking/sender"
+	"github.com/ava-labs/avalanchego/subnets"
+	"github.com/ava-labs/avalanchego/utils/set"
+)
+
+var (
+	_               simplex.Communication = (*Comm)(nil)
+	errNodeNotFound                       = errors.New("node not found in the validator list")
+)
+
+type Comm struct {
+	logger   simplex.Logger
+	subnetID ids.ID
+	chainID  ids.ID
+	// nodeID is this nodes ID
+	nodeID simplex.NodeID
+	// nodes are the IDs of all the nodes in the subnet
+	nodes []simplex.NodeID
+	// sender is used to send messages to other nodes
+	sender     sender.ExternalSender
+	msgBuilder message.OutboundMsgBuilder
+}
+
+func NewComm(config *Config) (*Comm, error) {
+	nodes := make([]simplex.NodeID, 0, len(config.Validators))
+
+	// grab all the nodes that are validators for the subnet
+	for _, vd := range config.Validators {
+		nodes = append(nodes, vd.NodeID[:])
+	}
+
+	if _, ok := config.Validators[config.Ctx.NodeID]; !ok {
+		config.Log.Warn("Node is not a validator for the subnet",
+			zap.String("nodeID", config.Ctx.NodeID.String()),
+			zap.String("chainID", config.Ctx.ChainID.String()),
+			zap.String("subnetID", config.Ctx.SubnetID.String()),
+		)
+		return nil, fmt.Errorf("%w could not find our node: %s", errNodeNotFound, config.Ctx.NodeID)
+	}
+
+	sortNodes(nodes)
+
+	c := &Comm{
+		subnetID:   config.Ctx.SubnetID,
+		nodes:      nodes,
+		nodeID:     config.Ctx.NodeID[:],
+		logger:     config.Log,
+		sender:     config.Sender,
+		msgBuilder: config.OutboundMsgBuilder,
+		chainID:    config.Ctx.ChainID,
+	}
+
+	return c, nil
+}
+
+// sortNodes sorts the nodes in place by their byte representations.
+func sortNodes(nodes []simplex.NodeID) {
+	slices.SortFunc(nodes, func(i, j simplex.NodeID) int {
+		return bytes.Compare(i, j)
+	})
+}
+
+func (c *Comm) ListNodes() []simplex.NodeID {
+	return c.nodes
+}
+
+func (c *Comm) SendMessage(msg *simplex.Message, destination simplex.NodeID) {
+	outboundMsg, err := c.simplexMessageToOutboundMessage(msg)
+	if err != nil {
+		c.logger.Error("Failed creating message", zap.Error(err))
+		return
+	}
+
+	dest, err := ids.ToNodeID(destination)
+	if err != nil {
+		c.logger.Error("Failed to convert destination NodeID", zap.Error(err))
+		return
+	}
+
+	c.sender.Send(outboundMsg, common.SendConfig{NodeIDs: set.Of(dest)}, c.subnetID, subnets.NoOpAllower)
+}
+
+func (c *Comm) Broadcast(msg *simplex.Message) {
+	for _, node := range c.nodes {
+		if node.Equals(c.nodeID) {
+			continue
+		}
+
+		c.SendMessage(msg, node)
+	}
+}
+
+func (c *Comm) simplexMessageToOutboundMessage(msg *simplex.Message) (message.OutboundMessage, error) {
+	var simplexMsg *p2p.Simplex
+	switch {
+	case msg.VerifiedBlockMessage != nil:
+		bytes, err := msg.VerifiedBlockMessage.VerifiedBlock.Bytes()
+		if err != nil {
+			return nil, fmt.Errorf("failed to serialize block: %w", err)
+		}
+		simplexMsg = newP2PSimplexBlockProposal(c.chainID, bytes, msg.VerifiedBlockMessage.Vote)
+	case msg.VoteMessage != nil:
+		simplexMsg = newP2PSimplexVote(c.chainID, msg.VoteMessage.Vote.BlockHeader, msg.VoteMessage.Signature)
+	case msg.EmptyVoteMessage != nil:
+		simplexMsg = newP2PSimplexEmptyVote(c.chainID, msg.EmptyVoteMessage.Vote.ProtocolMetadata, msg.EmptyVoteMessage.Signature)
+	case msg.FinalizeVote != nil:
+		simplexMsg = newP2PSimplexFinalizeVote(c.chainID, msg.FinalizeVote.Finalization.BlockHeader, msg.FinalizeVote.Signature)
+	case msg.Notarization != nil:
+		simplexMsg = newP2PSimplexNotarization(c.chainID, msg.Notarization.Vote.BlockHeader, msg.Notarization.QC.Bytes())
+	case msg.EmptyNotarization != nil:
+		simplexMsg = newP2PSimplexEmptyNotarization(c.chainID, msg.EmptyNotarization.Vote.ProtocolMetadata, msg.EmptyNotarization.QC.Bytes())
+	case msg.Finalization != nil:
+		simplexMsg = newP2PSimplexFinalization(c.chainID, msg.Finalization.Finalization.BlockHeader, msg.Finalization.QC.Bytes())
+	case msg.ReplicationRequest != nil:
+		simplexMsg = newP2PSimplexReplicationRequest(c.chainID, msg.ReplicationRequest.Seqs, msg.ReplicationRequest.LatestRound)
+	case msg.VerifiedReplicationResponse != nil:
+		msg, err := newP2PSimplexReplicationResponse(c.chainID, msg.VerifiedReplicationResponse.Data, msg.VerifiedReplicationResponse.LatestRound)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create replication response: %w", err)
+		}
+		simplexMsg = msg
+	}
+
+	return c.msgBuilder.SimplexMessage(simplexMsg)
+}

simplex/comm_test.go

@@ -0,0 +1,106 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package simplex
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ava-labs/simplex"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/message"
+	"github.com/ava-labs/avalanchego/message/messagemock"
+	"github.com/ava-labs/avalanchego/snow/networking/sender/sendermock"
+	"github.com/ava-labs/avalanchego/utils/constants"
+)
+
+var testSimplexMessage = simplex.Message{
+	VoteMessage: &simplex.Vote{
+		Vote: simplex.ToBeSignedVote{
+			BlockHeader: simplex.BlockHeader{
+				ProtocolMetadata: simplex.ProtocolMetadata{
+					Version: 1,
+					Epoch:   1,
+					Round:   1,
+					Seq:     1,
+				},
+			},
+		},
+		Signature: simplex.Signature{
+			Signer: []byte("dummy_node_id"),
+			Value:  []byte("dummy_signature"),
+		},
+	},
+}
+
+func TestCommSendMessage(t *testing.T) {
+	config := newEngineConfig(t, 1)
+
+	destinationNodeID := ids.GenerateTestNodeID()
+	ctrl := gomock.NewController(t)
+	sender := sendermock.NewExternalSender(ctrl)
+	mc, err := message.NewCreator(
+		prometheus.NewRegistry(),
+		constants.DefaultNetworkCompressionType,
+		10*time.Second,
+	)
+	require.NoError(t, err)
+
+	config.OutboundMsgBuilder = mc
+	config.Sender = sender
+
+	comm, err := NewComm(config)
+	require.NoError(t, err)
+
+	sender.EXPECT().Send(gomock.Any(), gomock.Any(), comm.subnetID, gomock.Any())
+
+	comm.SendMessage(&testSimplexMessage, destinationNodeID[:])
+}
+
+// TestCommBroadcast tests the Broadcast method sends to all nodes in the subnet
+// not including the sending node.
+func TestCommBroadcast(t *testing.T) {
+	config := newEngineConfig(t, 3)
+
+	ctrl := gomock.NewController(t)
+	sender := sendermock.NewExternalSender(ctrl)
+	mc, err := message.NewCreator(
+		prometheus.NewRegistry(),
+		constants.DefaultNetworkCompressionType,
+		10*time.Second,
+	)
+	require.NoError(t, err)
+
+	config.OutboundMsgBuilder = mc
+	config.Sender = sender
+
+	comm, err := NewComm(config)
+	require.NoError(t, err)
+
+	sender.EXPECT().Send(gomock.Any(), gomock.Any(), comm.subnetID, gomock.Any()).Times(2)
+
+	comm.Broadcast(&testSimplexMessage)
+}
+
+func TestCommFailsWithoutCurrentNode(t *testing.T) {
+	config := newEngineConfig(t, 3)
+
+	ctrl := gomock.NewController(t)
+	msgCreator := messagemock.NewOutboundMsgBuilder(ctrl)
+	sender := sendermock.NewExternalSender(ctrl)
+
+	config.OutboundMsgBuilder = msgCreator
+	config.Sender = sender
+
+	// set the curNode to a different nodeID than the one in the config
+	vdrs := generateTestNodes(t, 3)
+	config.Validators = newTestValidatorInfo(vdrs)
+
+	_, err := NewComm(config)
+	require.ErrorIs(t, err, errNodeNotFound)
+}

simplex/config.go

@@ -5,6 +5,8 @@ package simplex
 
 import (
 	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/message"
+	"github.com/ava-labs/avalanchego/snow/networking/sender"
 	"github.com/ava-labs/avalanchego/snow/validators"
 	"github.com/ava-labs/avalanchego/utils/logging"
 )
@@ -14,6 +16,9 @@ type Config struct {
 	Ctx SimplexChainContext
 	Log logging.Logger
 
+	Sender             sender.ExternalSender
+	OutboundMsgBuilder message.OutboundMsgBuilder
+
 	// Validators is a map of node IDs to their validator information.
 	// This tells the node about the current membership set, and should be consistent
 	// across all nodes in the subnet.
@@ -31,6 +36,9 @@ type SimplexChainContext struct {
 	// ChainID is the ID of the chain this context exists within.
 	ChainID ids.ID
 
+	// SubnetID is the ID of the subnet this context exists within.
+	SubnetID ids.ID
+
 	// NodeID is the ID of this node
 	NetworkID uint32
 }