fix(dot/network): Implement time based handle transaction

dot/network/config.go

@@ -41,7 +41,6 @@ const (
 	// DefaultDiscoveryInterval is the default interval for searching for DHT peers
 	DefaultDiscoveryInterval = time.Minute * 5
 
-	// defaultTxnBatchSize is the default size for the transaction batch
 	defaultTxnBatchSize = 100
 )
 
@@ -98,7 +97,7 @@ type Config struct {
 
 	batchSize int // internal option
 
-	// SlotDuration is slot duration to produce block in milliseconds
+	// SlotDuration is slot duration to produce block
 	SlotDuration time.Duration
 }
 

dot/network/host.go

@@ -391,7 +391,7 @@ func (h *host) closeProtocolStream(pID protocol.ID, p peer.ID) {
 			}
 			err := st.Close()
 			if err != nil {
-				logger.Tracef("Failed to close stream", "protocol", pID, "error", err)
+				logger.Tracef("Failed to close stream for protocol %s: %s", pID, err)
 			}
 		}
 	}

dot/network/transaction.go

@@ -20,6 +20,9 @@ var (
 	_ NotificationsMessage = &transactionHandshake{}
 )
 
+// txnBatchChTimeout is the timeout for adding a transaction to the batch processing channel
+const txnBatchChTimeout = time.Millisecond * 200
+
 // TransactionMessage is a network message that is sent to notify of new transactions entering the network
 type TransactionMessage struct {
 	Extrinsics []types.Extrinsic
@@ -107,58 +110,60 @@ func decodeTransactionHandshake(_ []byte) (Handshake, error) {
 	return &transactionHandshake{}, nil
 }
 
-func (s *Service) createBatchMessageHandler(txnBatchCh chan *BatchMessage) NotificationsMessageBatchHandler {
-	go func() {
-		protocolID := s.host.protocolID + transactionsID
-		ticker := time.NewTicker(s.cfg.SlotDuration)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-s.ctx.Done():
-				return
-			case <-ticker.C:
-				timeOut := time.NewTimer(s.cfg.SlotDuration / 3)
-				var completed bool
-				for !completed {
-					select {
-					case <-timeOut.C:
-						completed = true
-					case txnMsg := <-txnBatchCh:
-						propagate, err := s.handleTransactionMessage(txnMsg.peer, txnMsg.msg)
-						if err != nil {
-							s.host.closeProtocolStream(protocolID, txnMsg.peer)
-							continue
-						}
-
-						if s.noGossip || !propagate {
-							continue
-						}
-
-						if !s.gossip.hasSeen(txnMsg.msg) {
-							s.broadcastExcluding(s.notificationsProtocols[TransactionMsgType], txnMsg.peer, txnMsg.msg)
-						}
+func (s *Service) startTxnBatchProcessing(txnBatchCh chan *BatchMessage) {
+	protocolID := s.host.protocolID + transactionsID
+	ticker := time.NewTicker(s.cfg.SlotDuration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-ticker.C:
+			timeOut := time.NewTimer(s.cfg.SlotDuration / 3)
+			var completed bool
+			for !completed {
+				select {
+				case <-timeOut.C:
+					completed = true
+				case txnMsg := <-txnBatchCh:
+					propagate, err := s.handleTransactionMessage(txnMsg.peer, txnMsg.msg)
+					if err != nil {
+						s.host.closeProtocolStream(protocolID, txnMsg.peer)
+						continue
+					}
+
+					if s.noGossip || !propagate {
+						continue
+					}
+
+					if !s.gossip.hasSeen(txnMsg.msg) {
+						s.broadcastExcluding(s.notificationsProtocols[TransactionMsgType], txnMsg.peer, txnMsg.msg)
 					}
 				}
 			}
 		}
-	}()
+	}
+}
+
+func (s *Service) createBatchMessageHandler(txnBatchCh chan *BatchMessage) NotificationsMessageBatchHandler {
+	go s.startTxnBatchProcessing(txnBatchCh)
 
 	return func(peer peer.ID, msg NotificationsMessage) {
 		data := &BatchMessage{
 			msg:  msg,
 			peer: peer,
 		}
 
-		timeOut := time.NewTimer(time.Millisecond * 200)
+		timer := time.NewTimer(txnBatchChTimeout)
 
 		select {
 		case txnBatchCh <- data:
-			if !timeOut.Stop() {
-				<-timeOut.C
+			if !timer.Stop() {
+				<-timer.C
 			}
-		case <-timeOut.C:
-			logger.Debugf("transaction message not included into batch", "peer", peer.String(), "msg", msg.String())
+		case <-timer.C:
+			logger.Debugf("transaction message %s for peer %s not included into batch", msg, peer)
 		}
 	}
 }

dot/services.go

@@ -250,7 +250,7 @@ func createNetworkService(cfg *Config, stateSrvc *state.Service) (*network.Servi
 
 	slotDuration, err := stateSrvc.Epoch.GetSlotDuration()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("cannot get slot duration: %w", err)
 	}
 
 	// network service configuation