lib/runtime: use __heap_base as allocator offset

dot/network/block_announce_test.go

@@ -103,8 +103,8 @@ func TestHandleBlockAnnounceMessage(t *testing.T) {
 		Number: big.NewInt(10),
 	}
 
-	s.handleBlockAnnounceMessage(peerID, msg)
-	require.NotNil(t, s.syncQueue.isSyncing(peerID))
+	err := s.handleBlockAnnounceMessage(peerID, msg)
+	require.NoError(t, err)
 }
 
 func TestValidateBlockAnnounceHandshake(t *testing.T) {

dot/network/host.go

@@ -166,33 +166,10 @@ func (h *host) bootstrap() {
 // send writes the given message to the outbound message stream for the given
 // peer (gets the already opened outbound message stream or opens a new one).
 func (h *host) send(p peer.ID, sub protocol.ID, msg Message) (err error) {
-	encMsg, err := msg.Encode()
-	if err != nil {
-		return err
-	}
-
-	err = h.sendBytes(p, sub, encMsg)
-	if err != nil {
-		return err
-	}
-
-	logger.Trace(
-		"Sent message to peer",
-		"sub-protocol", sub,
-		"host", h.id(),
-		"peer", p,
-		"message", msg.String(),
-	)
-
-	return nil
-}
-
-func (h *host) sendBytes(p peer.ID, sub protocol.ID, msg []byte) (err error) {
 	// get outbound stream for given peer
 	s := h.getStream(p, sub)
 
 	// check if stream needs to be opened
-	// TODO: if stream has not been opened, begin handler for the newly opened stream
 	if s == nil {
 		logger.Debug("opening new stream with peer", "peer", p, "sub-protocol", sub)
 
@@ -211,11 +188,33 @@ func (h *host) sendBytes(p peer.ID, sub protocol.ID, msg []byte) (err error) {
 		)
 	}
 
-	msgLen := uint64(len(msg))
+	err = h.writeToStream(s, msg)
+	if err != nil {
+		return err
+	}
+
+	logger.Trace(
+		"Sent message to peer",
+		"sub-protocol", sub,
+		"host", h.id(),
+		"peer", p,
+		"message", msg.String(),
+	)
+
+	return nil
+}
+
+func (h *host) writeToStream(s libp2pnetwork.Stream, msg Message) error {
+	encMsg, err := msg.Encode()
+	if err != nil {
+		return err
+	}
+
+	msgLen := uint64(len(encMsg))
 	lenBytes := uint64ToLEB128(msgLen)
-	msg = append(lenBytes, msg...)
+	encMsg = append(lenBytes, encMsg...)
 
-	_, err = s.Write(msg)
+	_, err = s.Write(encMsg)
 	return err
 }
 

dot/network/host_test.go

@@ -23,70 +23,12 @@ import (
 	"time"
 
 	"github.com/ChainSafe/gossamer/lib/utils"
-	log "github.com/ChainSafe/log15"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-kad-dht/dual"
 	kbucket "github.com/libp2p/go-libp2p-kbucket"
 	"github.com/stretchr/testify/require"
 )
 
-func TestMessageSize(t *testing.T) {
-	size := 100000
-	msg := make([]byte, size)
-	msg[0] = 77
-
-	nodes := make([]*Service, 2)
-	errs := make([]chan error, 2)
-
-	for i := range nodes {
-		errCh := make(chan error)
-
-		config := &Config{
-			Port:        7000 + uint32(i),
-			RandSeed:    1 + int64(i),
-			NoBootstrap: true,
-			NoMDNS:      true,
-			ErrChan:     errCh,
-			LogLvl:      log.LvlTrace,
-		}
-		node := createTestService(t, config)
-		defer node.Stop()
-		nodes[i] = node
-		errs[i] = errCh
-		handler := newTestStreamHandler(testBlockRequestMessageDecoder)
-		node.host.registerStreamHandler("", handler.handleStream)
-	}
-
-	addrs := nodes[0].host.multiaddrs()
-	ainfo, err := peer.AddrInfoFromP2pAddr(addrs[1])
-	require.NoError(t, err)
-
-	for i, n := range nodes {
-		if i == 0 {
-			// connect other nodes to first node
-			continue
-		}
-
-		err = n.host.connect(*ainfo)
-		require.NoError(t, err, i)
-	}
-
-	err = nodes[0].host.sendBytes(nodes[1].host.id(), "", msg)
-	require.NoError(t, err)
-	time.Sleep(time.Second)
-	err = nodes[1].host.sendBytes(nodes[0].host.id(), "", msg)
-	require.NoError(t, err)
-	time.Sleep(time.Second)
-
-	select {
-	case err := <-errs[0]:
-		t.Fatal("got error", err)
-	case err := <-errs[1]:
-		t.Fatal("got error", err)
-	case <-time.After(time.Millisecond * 100):
-	}
-}
-
 // test host connect method
 func TestConnect(t *testing.T) {
 	basePathA := utils.NewTestBasePath(t, "nodeA")

dot/network/light.go

@@ -6,7 +6,6 @@ import (
 	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/ChainSafe/gossamer/lib/common/optional"
 	"github.com/ChainSafe/gossamer/lib/scale"
-	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 // Pair is a pair of arbitrary bytes.
@@ -263,18 +262,18 @@ func (rh *RemoteHeaderResponse) String() string {
 	return fmt.Sprintf("Header =%s Proof =%s", rh.Header, string(rh.proof))
 }
 
-func remoteCallResp(peer peer.ID, req *RemoteCallRequest) (*RemoteCallResponse, error) {
+func remoteCallResp(req *RemoteCallRequest) (*RemoteCallResponse, error) {
 	return &RemoteCallResponse{}, nil
 }
-func remoteChangeResp(peer peer.ID, req *RemoteChangesRequest) (*RemoteChangesResponse, error) {
+func remoteChangeResp(req *RemoteChangesRequest) (*RemoteChangesResponse, error) {
 	return &RemoteChangesResponse{}, nil
 }
-func remoteHeaderResp(peer peer.ID, req *RemoteHeaderRequest) (*RemoteHeaderResponse, error) {
+func remoteHeaderResp(req *RemoteHeaderRequest) (*RemoteHeaderResponse, error) {
 	return &RemoteHeaderResponse{}, nil
 }
-func remoteReadChildResp(peer peer.ID, req *RemoteReadChildRequest) (*RemoteReadResponse, error) {
+func remoteReadChildResp(req *RemoteReadChildRequest) (*RemoteReadResponse, error) {
 	return &RemoteReadResponse{}, nil
 }
-func remoteReadResp(peer peer.ID, req *RemoteReadRequest) (*RemoteReadResponse, error) {
+func remoteReadResp(req *RemoteReadRequest) (*RemoteReadResponse, error) {
 	return &RemoteReadResponse{}, nil
 }

dot/network/light_test.go

@@ -2,6 +2,7 @@ package network
 
 import (
 	"testing"
+	"time"
 
 	"github.com/ChainSafe/gossamer/lib/utils"
 	"github.com/libp2p/go-libp2p-core/peer"
@@ -45,23 +46,41 @@ func TestDecodeLightMessage(t *testing.T) {
 }
 
 func TestHandleLightMessage_Response(t *testing.T) {
-	basePath := utils.NewTestBasePath(t, "nodeA")
-	defer utils.RemoveTestDir(t)
-
 	config := &Config{
-		BasePath:    basePath,
+		BasePath:    utils.NewTestBasePath(t, "nodeA"),
 		Port:        7001,
 		RandSeed:    1,
 		NoBootstrap: true,
 		NoMDNS:      true,
 	}
 	s := createTestService(t, config)
 
-	peerID := peer.ID("noot")
+	configB := &Config{
+		BasePath:    utils.NewTestBasePath(t, "nodeB"),
+		Port:        7002,
+		RandSeed:    2,
+		NoBootstrap: true,
+		NoMDNS:      true,
+	}
+	b := createTestService(t, configB)
+
+	addrInfosB, err := b.host.addrInfos()
+	require.NoError(t, err)
+
+	err = s.host.connect(*addrInfosB[0])
+	// retry connect if "failed to dial" error
+	if failedToDial(err) {
+		time.Sleep(TestBackoffTimeout)
+		err = s.host.connect(*addrInfosB[0])
+	}
+	require.NoError(t, err)
+
+	stream, err := s.host.h.NewStream(s.ctx, b.host.id(), s.host.protocolID+lightID)
+	require.NoError(t, err)
 
 	// Testing empty request
 	msg := &LightRequest{}
-	err := s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.NoError(t, err)
 
 	expectedErr := "failed to find any peer in table"
@@ -70,34 +89,34 @@ func TestHandleLightMessage_Response(t *testing.T) {
 	msg = &LightRequest{
 		RmtCallRequest: &RemoteCallRequest{},
 	}
-	err = s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.Error(t, err, expectedErr, msg.String())
 
 	// Testing remoteHeaderResp()
 	msg = &LightRequest{
 		RmtHeaderRequest: &RemoteHeaderRequest{},
 	}
-	err = s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.Error(t, err, expectedErr, msg.String())
 
 	// Testing remoteChangeResp()
 	msg = &LightRequest{
 		RmtChangesRequest: &RemoteChangesRequest{},
 	}
-	err = s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.Error(t, err, expectedErr, msg.String())
 
 	// Testing remoteReadResp()
 	msg = &LightRequest{
 		RmtReadRequest: &RemoteReadRequest{},
 	}
-	err = s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.Error(t, err, expectedErr, msg.String())
 
 	// Testing remoteReadChildResp()
 	msg = &LightRequest{
 		RmtReadChildRequest: &RemoteReadChildRequest{},
 	}
-	err = s.handleLightMsg(peerID, msg)
+	err = s.handleLightMsg(stream, msg)
 	require.Error(t, err, expectedErr, msg.String())
 }

dot/network/message.go

@@ -150,7 +150,7 @@ func (bm *BlockRequestMessage) Decode(in []byte) error {
 	case *pb.BlockRequest_Number:
 		startingBlock, err = variadic.NewUint64OrHash(binary.LittleEndian.Uint64(from.Number))
 	default:
-		err = errors.New("invalid FromBlock")
+		err = errors.New("invalid StartingBlock")
 	}
 
 	if err != nil {

dot/network/notifications.go

@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"sync"
 
+	libp2pnetwork "github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
 )
@@ -80,14 +81,15 @@ func createDecoder(info *notificationsProtocol, handshakeDecoder HandshakeDecode
 }
 
 func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol, handshakeValidator HandshakeValidator, messageHandler NotificationsMessageHandler) messageHandler {
-	return func(peer peer.ID, m Message) error {
+	return func(stream libp2pnetwork.Stream, m Message) error {
 		if m == nil || info == nil || handshakeValidator == nil || messageHandler == nil {
 			return nil
 		}
 
 		var (
-			ok  bool
-			msg NotificationsMessage
+			ok   bool
+			msg  NotificationsMessage
+			peer = stream.Conn().RemotePeer()
 		)
 
 		if msg, ok = m.(NotificationsMessage); !ok {

dot/network/notifications_test.go

@@ -131,6 +131,9 @@ func TestCreateNotificationsMessageHandler_BlockAnnounce(t *testing.T) {
 	}
 	require.NoError(t, err)
 
+	stream, err := s.host.h.NewStream(s.ctx, b.host.id(), s.host.protocolID+blockAnnounceID)
+	require.NoError(t, err)
+
 	// create info and handler
 	info := &notificationsProtocol{
 		subProtocol:   blockAnnounceID,
@@ -148,9 +151,8 @@ func TestCreateNotificationsMessageHandler_BlockAnnounce(t *testing.T) {
 		Number: big.NewInt(10),
 	}
 
-	err = handler(testPeerID, msg)
+	err = handler(stream, msg)
 	require.NoError(t, err)
-	require.NotNil(t, s.syncQueue.isSyncing(testPeerID))
 }
 
 func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T) {
@@ -185,6 +187,20 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
 	// don't set handshake data ie. this stream has just been opened
 	testPeerID := b.host.id()
 
+	// connect nodes
+	addrInfosB, err := b.host.addrInfos()
+	require.NoError(t, err)
+
+	err = s.host.connect(*addrInfosB[0])
+	if failedToDial(err) {
+		time.Sleep(TestBackoffTimeout)
+		err = s.host.connect(*addrInfosB[0])
+	}
+	require.NoError(t, err)
+
+	stream, err := s.host.h.NewStream(s.ctx, b.host.id(), s.host.protocolID+blockAnnounceID)
+	require.NoError(t, err)
+
 	// try invalid handshake
 	testHandshake := &BlockAnnounceHandshake{
 		Roles:           4,
@@ -193,22 +209,11 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
 		GenesisHash:     common.Hash{2},
 	}
 
-	err := handler(testPeerID, testHandshake)
+	err = handler(stream, testHandshake)
 	require.Equal(t, errCannotValidateHandshake, err)
 	require.True(t, info.handshakeData[testPeerID].received)
 	require.False(t, info.handshakeData[testPeerID].validated)
 
-	// connect nodes
-	addrInfosB, err := b.host.addrInfos()
-	require.NoError(t, err)
-
-	err = s.host.connect(*addrInfosB[0])
-	if failedToDial(err) {
-		time.Sleep(TestBackoffTimeout)
-		err = s.host.connect(*addrInfosB[0])
-	}
-	require.NoError(t, err)
-
 	// try valid handshake
 	testHandshake = &BlockAnnounceHandshake{
 		Roles:           4,
@@ -217,7 +222,7 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
 		GenesisHash:     s.blockState.GenesisHash(),
 	}
 
-	err = handler(testPeerID, testHandshake)
+	err = handler(stream, testHandshake)
 	require.NoError(t, err)
 	require.True(t, info.handshakeData[testPeerID].received)
 	require.True(t, info.handshakeData[testPeerID].validated)