Comm Component for Simplex

go.mod

@@ -87,7 +87,7 @@ require (
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/VictoriaMetrics/fastcache v1.12.1 // indirect
 	github.com/ava-labs/firewood-go-ethhash/ffi v0.0.8 // indirect
-	github.com/ava-labs/simplex v0.0.0-20250626192006-220e6aeacdc1
+	github.com/ava-labs/simplex v0.0.0-20250715173145-e4fe035cb9b2
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.10.0 // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.3.2 // indirect

go.sum

@@ -78,8 +78,8 @@ github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60 h1:EL
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60/go.mod h1:/7qKobTfbzBu7eSTVaXMTr56yTYk4j2Px6/8G+idxHo=
 github.com/ava-labs/libevm v1.13.14-0.3.0.rc.1 h1:vBMYo+Iazw0rGTr+cwjkBdh5eadLPlv4ywI4lKye3CA=
 github.com/ava-labs/libevm v1.13.14-0.3.0.rc.1/go.mod h1:+Iol+sVQ1KyoBsHf3veyrBmHCXr3xXRWq6ZXkgVfNLU=
-github.com/ava-labs/simplex v0.0.0-20250626192006-220e6aeacdc1 h1:ipeWExRrhYF7DZ/bcigoQrzo3vZWNZrFx8W+Yg2sJ2Q=
-github.com/ava-labs/simplex v0.0.0-20250626192006-220e6aeacdc1/go.mod h1:GVzumIo3zR23/qGRN2AdnVkIPHcKMq/D89EGWZfMGQ0=
+github.com/ava-labs/simplex v0.0.0-20250715173145-e4fe035cb9b2 h1:PZ5PMEDkTbd6NLNiwKWV8nz7QvAM+QC9Rj3/NrL9ICA=
+github.com/ava-labs/simplex v0.0.0-20250715173145-e4fe035cb9b2/go.mod h1:GVzumIo3zR23/qGRN2AdnVkIPHcKMq/D89EGWZfMGQ0=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

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/block.go

@@ -67,7 +67,7 @@ type blockDeserializer struct {
 	parser block.Parser
 }
 
-func (d *blockDeserializer) DeserializeBlock(bytes []byte) (simplex.Block, error) {
+func (d *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte) (simplex.Block, error) {
 	var canotoBlock canotoSimplexBlock
 
 	if err := canotoBlock.UnmarshalCanoto(bytes); err != nil {
@@ -79,7 +79,7 @@ func (d *blockDeserializer) DeserializeBlock(bytes []byte) (simplex.Block, error
 		return nil, fmt.Errorf("failed to parse protocol metadata: %w", err)
 	}
 
-	vmblock, err := d.parser.ParseBlock(context.TODO(), canotoBlock.InnerBlock)
+	vmblock, err := d.parser.ParseBlock(ctx, canotoBlock.InnerBlock)
 	if err != nil {
 		return nil, err
 	}

simplex/block_test.go

@@ -21,7 +21,7 @@ import (
 
 func TestBlockSerialization(t *testing.T) {
 	unexpectedBlockBytes := errors.New("unexpected block bytes")
-
+	ctx := context.Background()
 	testBlock := snowmantest.BuildChild(snowmantest.Genesis)
 
 	b := &Block{
@@ -88,7 +88,7 @@ func TestBlockSerialization(t *testing.T) {
 			}
 
 			// Deserialize the block
-			deserializedBlock, err := deserializer.DeserializeBlock(tt.blockBytes)
+			deserializedBlock, err := deserializer.DeserializeBlock(ctx, tt.blockBytes)
 			require.ErrorIs(t, err, tt.expectedError)
 
 			if tt.expectedError == nil {

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,138 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package simplex
+
+import (
+	"errors"
+	"fmt"
+
+	"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
+	// broadcastNodes are the nodes that should receive broadcast messages
+	broadcastNodes set.Set[ids.NodeID]
+	// allNodes are the IDs of all the nodes in the subnet
+	allNodes []simplex.NodeID
+
+	// sender is used to send messages to other nodes
+	sender     sender.ExternalSender
+	msgBuilder message.OutboundMsgBuilder
+}
+
+func NewComm(config *Config) (*Comm, error) {
+	if _, ok := config.Validators[config.Ctx.NodeID]; !ok {
+		config.Log.Warn("Node is not a validator for the subnet",
+			zap.Stringer("nodeID", config.Ctx.NodeID),
+			zap.Stringer("chainID", config.Ctx.ChainID),
+			zap.Stringer("subnetID", config.Ctx.SubnetID),
+		)
+		return nil, fmt.Errorf("our %w: %s", errNodeNotFound, config.Ctx.NodeID)
+	}
+
+	broadcastNodes := set.NewSet[ids.NodeID](len(config.Validators) - 1)
+	allNodes := make([]simplex.NodeID, 0, len(config.Validators))
+	// grab all the nodes that are validators for the subnet
+	for _, vd := range config.Validators {
+		allNodes = append(allNodes, vd.NodeID[:])
+		if vd.NodeID == config.Ctx.NodeID {
+			continue // skip our own node ID
+		}
+
+		broadcastNodes.Add(vd.NodeID)
+	}
+
+	return &Comm{
+		subnetID:       config.Ctx.SubnetID,
+		broadcastNodes: broadcastNodes,
+		allNodes:       allNodes,
+		logger:         config.Log,
+		sender:         config.Sender,
+		msgBuilder:     config.OutboundMsgBuilder,
+		chainID:        config.Ctx.ChainID,
+	}, nil
+}
+
+func (c *Comm) Nodes() []simplex.NodeID {
+	return c.allNodes
+}
+
+func (c *Comm) Send(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) {
+	outboundMsg, err := c.simplexMessageToOutboundMessage(msg)
+	if err != nil {
+		c.logger.Error("Failed creating message", zap.Error(err))
+		return
+	}
+
+	c.sender.Send(outboundMsg, common.SendConfig{NodeIDs: c.broadcastNodes}, c.subnetID, subnets.NoOpAllower)
+}
+
+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 = newBlockProposal(c.chainID, bytes, msg.VerifiedBlockMessage.Vote)
+	case msg.VoteMessage != nil:
+		simplexMsg = newVote(c.chainID, msg.VoteMessage)
+	case msg.EmptyVoteMessage != nil:
+		simplexMsg = newEmptyVote(c.chainID, msg.EmptyVoteMessage)
+	case msg.FinalizeVote != nil:
+		simplexMsg = newFinalizeVote(c.chainID, msg.FinalizeVote)
+	case msg.Notarization != nil:
+		simplexMsg = newNotarization(c.chainID, msg.Notarization)
+	case msg.EmptyNotarization != nil:
+		simplexMsg = newEmptyNotarization(c.chainID, msg.EmptyNotarization)
+	case msg.Finalization != nil:
+		simplexMsg = newFinalization(c.chainID, msg.Finalization)
+	case msg.ReplicationRequest != nil:
+		simplexMsg = newReplicationRequest(c.chainID, msg.ReplicationRequest)
+	case msg.VerifiedReplicationResponse != nil:
+		msg, err := newReplicationResponse(c.chainID, msg.VerifiedReplicationResponse)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create replication response: %w", err)
+		}
+		simplexMsg = msg
+	default:
+		return nil, fmt.Errorf("unknown message type: %+v", msg)
+	}
+
+	return c.msgBuilder.SimplexMessage(simplexMsg)
+}

simplex/comm_test.go

@@ -0,0 +1,130 @@
+// 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/snow/engine/common"
+	"github.com/ava-labs/avalanchego/snow/networking/sender/sendermock"
+	"github.com/ava-labs/avalanchego/utils/constants"
+	"github.com/ava-labs/avalanchego/utils/set"
+)
+
+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)
+
+	outboundMsg, err := mc.SimplexMessage(newVote(config.Ctx.ChainID, testSimplexMessage.VoteMessage))
+	require.NoError(t, err)
+	expectedSendConfig := common.SendConfig{
+		NodeIDs: set.Of(destinationNodeID),
+	}
+	sender.EXPECT().Send(outboundMsg, expectedSendConfig, comm.subnetID, gomock.Any())
+
+	comm.Send(&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)
+	outboundMsg, err := mc.SimplexMessage(newVote(config.Ctx.ChainID, testSimplexMessage.VoteMessage))
+	require.NoError(t, err)
+	nodes := make([]ids.NodeID, 0, len(comm.Nodes()))
+	for _, node := range comm.Nodes() {
+		if node.Equals(config.Ctx.NodeID[:]) {
+			continue // skip the sending node
+		}
+		nodes = append(nodes, ids.NodeID(node))
+	}
+
+	expectedSendConfig := common.SendConfig{
+		NodeIDs: set.Of(nodes...),
+	}
+
+	sender.EXPECT().Send(outboundMsg, expectedSendConfig, comm.subnetID, gomock.Any())
+
+	comm.Broadcast(&testSimplexMessage)
+}
+
+func TestCommFailsWithoutCurrentNode(t *testing.T) {
+	config := newEngineConfig(t, 3)
+
+	ctrl := gomock.NewController(t)
+	mc, err := message.NewCreator(
+		prometheus.NewRegistry(),
+		constants.DefaultNetworkCompressionType,
+		10*time.Second,
+	)
+	require.NoError(t, err)
+	sender := sendermock.NewExternalSender(ctrl)
+
+	config.OutboundMsgBuilder = mc
+	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)
+}