chore (pkg/babe): Integrate scale into babe and transaction libraries

chain/dev/genesis.json

@@ -3,7 +3,12 @@
     "id": "dev",
     "chainType": "Local",
     "bootNodes": [],
-    "telemetryEndpoints": null,
+    "telemetryEndpoints": [
+    [
+      "wss://telemetry.polkadot.io/submit/",
+      0
+    ]
+  ],
     "protocolId": "/gossamer/dev/0",
     "genesis": {
         "raw": {
@@ -32,4 +37,4 @@
     "forkBlocks": null,
     "badBlocks": null,
     "consensusEngine": ""
-}
+}

chain/gssmr/genesis.json

@@ -3,6 +3,12 @@
     "id": "gssmr",
     "chainType": "Local",
     "bootNodes": [],
+    "telemetryEndpoints": [
+    [
+      "wss://telemetry.polkadot.io/submit/",
+      0
+    ]
+  ],
     "protocolId": "/gossamer/gssmr/0",
     "genesis": {
         "raw": {
@@ -40,4 +46,4 @@
     "forkBlocks": null,
     "badBlocks": null,
     "consensusEngine": ""
-}
+}

dot/core/messages.go

@@ -57,3 +57,8 @@ func (s *Service) HandleTransactionMessage(msg *network.TransactionMessage) (boo
 
 	return len(msg.Extrinsics) > 0, nil
 }
+
+// TransactionsCount returns number for pending transactions in pool
+func (s *Service) TransactionsCount() int {
+	return len(s.transactionState.PendingInPool())
+}

dot/core/messages_test.go

@@ -21,7 +21,7 @@ import (
 	"testing"
 	"time"
 
-	. "github.com/ChainSafe/gossamer/dot/core/mocks"
+	. "github.com/ChainSafe/gossamer/dot/core/mocks" // nolint
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
@@ -38,7 +38,7 @@ import (
 
 func TestService_ProcessBlockAnnounceMessage(t *testing.T) {
 	// TODO: move to sync package
-	net := new(MockNetwork)
+	net := new(MockNetwork) // nolint
 
 	cfg := &Config{
 		Network:  net,
@@ -136,7 +136,7 @@ func TestService_HandleTransactionMessage(t *testing.T) {
 	ks := keystore.NewGlobalKeystore()
 	ks.Acco.Insert(kp)
 
-	bp := new(MockBlockProducer)
+	bp := new(MockBlockProducer) // nolint
 	blockC := make(chan types.Block)
 	bp.On("GetBlockChannel", nil).Return(blockC)
 

dot/network/discovery.go

@@ -75,13 +75,15 @@ func (d *discovery) start() error {
 		// get all currently connected peers and use them to bootstrap the DHT
 		peers := d.h.Network().Peers()
 
+		t := time.NewTicker(startDHTTimeout)
+		defer t.Stop()
 		for {
 			if len(peers) > 0 {
 				break
 			}
 
 			select {
-			case <-time.After(startDHTTimeout):
+			case <-t.C:
 				logger.Debug("no peers yet, waiting to start DHT...")
 				// wait for peers to connect before starting DHT, otherwise DHT bootstrap nodes
 				// will be empty and we will fail to fill the routing table
@@ -169,11 +171,13 @@ func (d *discovery) advertise() {
 }
 
 func (d *discovery) checkPeerCount() {
+	t := time.NewTicker(connectToPeersTimeout)
+	defer t.Stop()
 	for {
 		select {
 		case <-d.ctx.Done():
 			return
-		case <-time.After(connectToPeersTimeout):
+		case <-t.C:
 			if len(d.h.Network().Peers()) > d.minPeers {
 				continue
 			}

dot/network/service.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"math/big"
 	"os"
 	"sync"
 	"time"
@@ -315,11 +316,12 @@ main:
 
 		case <-ticker.C:
 			o := s.host.bwc.GetBandwidthTotals()
-			err := telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-				telemetry.NewKeyValue("bandwidth_download", o.RateIn),
-				telemetry.NewKeyValue("bandwidth_upload", o.RateOut),
-				telemetry.NewKeyValue("msg", "system.interval"),
-				telemetry.NewKeyValue("peers", s.host.peerCount())))
+			err := telemetry.GetInstance().SendMessage(telemetry.NewBandwidthTM(o.RateIn, o.RateOut, s.host.peerCount()))
+			if err != nil {
+				logger.Debug("problem sending system.interval telemetry message", "error", err)
+			}
+
+			err = telemetry.GetInstance().SendMessage(telemetry.NewNetworkStateTM(s.host.h, s.Peers()))
 			if err != nil {
 				logger.Debug("problem sending system.interval telemetry message", "error", err)
 			}
@@ -333,19 +335,22 @@ func (s *Service) sentBlockIntervalTelemetry() {
 		if err != nil {
 			continue
 		}
+		bestHash := best.Hash()
+
 		finalized, err := s.blockState.GetFinalizedHeader(0, 0) //nolint
 		if err != nil {
 			continue
 		}
-
-		err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-			telemetry.NewKeyValue("best", best.Hash().String()),
-			telemetry.NewKeyValue("finalized_hash", finalized.Hash().String()), //nolint
-			telemetry.NewKeyValue("finalized_height", finalized.Number),        //nolint
-			telemetry.NewKeyValue("height", best.Number),
-			telemetry.NewKeyValue("msg", "system.interval"),
-			telemetry.NewKeyValue("txcount", 0),                // todo (ed) determine where to get tx count
-			telemetry.NewKeyValue("used_state_cache_size", 0))) // todo (ed) determine where to get used_state_cache_size
+		finalizedHash := finalized.Hash()
+
+		err = telemetry.GetInstance().SendMessage(telemetry.NewBlockIntervalTM(
+			&bestHash,
+			best.Number,
+			&finalizedHash,
+			finalized.Number,
+			big.NewInt(int64(s.transactionHandler.TransactionsCount())),
+			big.NewInt(0), // todo (ed) determine where to get used_state_cache_size
+		))
 		if err != nil {
 			logger.Debug("problem sending system.interval telemetry message", "error", err)
 		}

dot/network/service_test.go

@@ -84,6 +84,7 @@ func createTestService(t *testing.T, cfg *Config) (srvc *Service) {
 	if cfg.TransactionHandler == nil {
 		mocktxhandler := &MockTransactionHandler{}
 		mocktxhandler.On("HandleTransactionMessage", mock.AnythingOfType("*TransactionMessage")).Return(nil)
+		mocktxhandler.On("TransactionsCount").Return(0)
 		cfg.TransactionHandler = mocktxhandler
 	}
 

dot/network/state.go

@@ -56,4 +56,5 @@ type Syncer interface {
 // TransactionHandler is the interface used by the transactions sub-protocol
 type TransactionHandler interface {
 	HandleTransactionMessage(*TransactionMessage) (bool, error)
+	TransactionsCount() int
 }

dot/network/sync.go

@@ -88,7 +88,9 @@ const (
 	badPeerThreshold       int    = -2
 	protectedPeerThreshold int    = 7
 
-	defaultSlotDuration = time.Second * 6
+	defaultSlotDuration                = time.Second * 6
+	defaultHandleResponseQueueDuration = time.Second
+	defaultPrunePeersDuration          = time.Second * 30
 )
 
 var (
@@ -132,26 +134,30 @@ type syncQueue struct {
 	goal               int64 // goal block number we are trying to sync to
 	currStart, currEnd int64 // the start and end of the BlockResponse we are currently handling; 0 and 0 if we are not currently handling any
 
-	benchmarker *syncBenchmarker
+	benchmarker                 *syncBenchmarker
+	handleResponseQueueDuration time.Duration
+	prunePeersDuration          time.Duration
 }
 
 func newSyncQueue(s *Service) *syncQueue {
 	ctx, cancel := context.WithCancel(s.ctx)
 
 	return &syncQueue{
-		s:                        s,
-		slotDuration:             defaultSlotDuration,
-		ctx:                      ctx,
-		cancel:                   cancel,
-		peerScore:                new(sync.Map),
-		requestData:              new(sync.Map),
-		requestDataByHash:        new(sync.Map),
-		justificationRequestData: new(sync.Map),
-		requestCh:                make(chan *syncRequest, blockRequestBufferSize),
-		responses:                []*types.BlockData{},
-		responseCh:               make(chan []*types.BlockData, blockResponseBufferSize),
-		benchmarker:              newSyncBenchmarker(),
-		buf:                      make([]byte, maxBlockResponseSize),
+		s:                           s,
+		slotDuration:                defaultSlotDuration,
+		ctx:                         ctx,
+		cancel:                      cancel,
+		peerScore:                   new(sync.Map),
+		requestData:                 new(sync.Map),
+		requestDataByHash:           new(sync.Map),
+		justificationRequestData:    new(sync.Map),
+		requestCh:                   make(chan *syncRequest, blockRequestBufferSize),
+		responses:                   []*types.BlockData{},
+		responseCh:                  make(chan []*types.BlockData, blockResponseBufferSize),
+		benchmarker:                 newSyncBenchmarker(),
+		buf:                         make([]byte, maxBlockResponseSize),
+		handleResponseQueueDuration: defaultHandleResponseQueueDuration,
+		prunePeersDuration:          defaultPrunePeersDuration,
 	}
 }
 
@@ -176,10 +182,12 @@ func (q *syncQueue) syncAtHead() {
 	q.s.syncer.SetSyncing(true)
 	q.s.noGossip = true // don't gossip messages until we're at the head
 
+	t := time.NewTicker(q.slotDuration * 2)
+	defer t.Stop()
 	for {
 		select {
 		// sleep for average block time TODO: make this configurable from slot duration
-		case <-time.After(q.slotDuration * 2):
+		case <-t.C:
 		case <-q.ctx.Done():
 			return
 		}
@@ -214,9 +222,11 @@ func (q *syncQueue) syncAtHead() {
 }
 
 func (q *syncQueue) handleResponseQueue() {
+	t := time.NewTicker(q.handleResponseQueueDuration)
+	defer t.Stop()
 	for {
 		select {
-		case <-time.After(time.Second):
+		case <-t.C:
 		case <-q.ctx.Done():
 			return
 		}
@@ -260,9 +270,11 @@ func (q *syncQueue) handleResponseQueue() {
 
 // prune peers with low score and connect to new peers
 func (q *syncQueue) prunePeers() {
+	t := time.NewTicker(q.prunePeersDuration)
+	defer t.Stop()
 	for {
 		select {
-		case <-time.After(time.Second * 30):
+		case <-t.C:
 		case <-q.ctx.Done():
 			return
 		}

dot/network/test_helpers.go

@@ -59,6 +59,7 @@ func NewMockSyncer() *MockSyncer {
 func NewMockTransactionHandler() *MockTransactionHandler {
 	mocktxhandler := new(MockTransactionHandler)
 	mocktxhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(nil)
+	mocktxhandler.On("TransactionsCount").Return(0)
 	return mocktxhandler
 }
 

dot/network/transaction_test.go

@@ -56,6 +56,7 @@ func TestHandleTransactionMessage(t *testing.T) {
 	basePath := utils.NewTestBasePath(t, "nodeA")
 	mockhandler := &MockTransactionHandler{}
 	mockhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(true, nil)
+	mockhandler.On("TransactionsCount").Return(0)
 
 	config := &Config{
 		BasePath:           basePath,

dot/node.go

@@ -345,17 +345,16 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 	}
 
 	telemetry.GetInstance().AddConnections(gd.TelemetryEndpoints)
-
-	err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-		telemetry.NewKeyValue("authority", cfg.Core.GrandpaAuthority),
-		telemetry.NewKeyValue("chain", sysSrvc.ChainName()),
-		telemetry.NewKeyValue("genesis_hash", stateSrvc.Block.GenesisHash().String()),
-		telemetry.NewKeyValue("implementation", sysSrvc.SystemName()),
-		telemetry.NewKeyValue("msg", "system.connected"),
-		telemetry.NewKeyValue("name", cfg.Global.Name),
-		telemetry.NewKeyValue("network_id", networkSrvc.NetworkState().PeerID),
-		telemetry.NewKeyValue("startup_time", strconv.FormatInt(time.Now().UnixNano(), 10)),
-		telemetry.NewKeyValue("version", sysSrvc.SystemVersion())))
+	genesisHash := stateSrvc.Block.GenesisHash()
+	err = telemetry.GetInstance().SendMessage(telemetry.NewSystemConnectedTM(
+		cfg.Core.GrandpaAuthority,
+		sysSrvc.ChainName(),
+		&genesisHash,
+		sysSrvc.SystemName(),
+		cfg.Global.Name,
+		networkSrvc.NetworkState().PeerID,
+		strconv.FormatInt(time.Now().UnixNano(), 10),
+		sysSrvc.SystemVersion()))
 	if err != nil {
 		logger.Debug("problem sending system.connected telemetry message", "err", err)
 	}

dot/sync/syncer.go

@@ -346,13 +346,13 @@ func (s *Service) handleBlock(block *types.Block) error {
 
 	logger.Debug("🔗 imported block", "number", block.Header.Number, "hash", block.Header.Hash())
 
-	err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage( // nolint
-		telemetry.NewKeyValue("best", block.Header.Hash().String()),
-		telemetry.NewKeyValue("height", block.Header.Number.Uint64()),
-		telemetry.NewKeyValue("msg", "block.import"),
-		telemetry.NewKeyValue("origin", "NetworkInitialSync")))
+	blockHash := block.Header.Hash()
+	err = telemetry.GetInstance().SendMessage(telemetry.NewBlockImportTM(
+		&blockHash,
+		block.Header.Number,
+		"NetworkInitialSync"))
 	if err != nil {
-		logger.Trace("problem sending block.import telemetry message", "error", err)
+		logger.Debug("problem sending block.import telemetry message", "error", err)
 	}
 
 	return nil