chore: rollback modification to network pkg

dot/network/service.go

@@ -31,13 +31,11 @@ const (
 
 	// the following are sub-protocols used by the node
 	syncID          = "/sync/2"
-	warpSyncID      = "/sync/warp"
 	lightID         = "/light/2"
 	blockAnnounceID = "/block-announces/1"
 	transactionsID  = "/transactions/1"
 
-	warpSyncMaxResponseSize = 16 * 1024 * 1024
-	maxMessageSize          = 1024 * 64 // 64kb for now
+	maxMessageSize = 1024 * 64 // 64kb for now
 )
 
 var (
@@ -249,8 +247,6 @@ func (s *Service) Start() error {
 	}
 
 	s.host.registerStreamHandler(s.host.protocolID+syncID, s.handleSyncStream)
-	// TODO: enable this protocol to receive request from other nodes
-	//s.host.registerStreamHandler(s.host.protocolID+warpSync, s.handleSyncStream)
 	s.host.registerStreamHandler(s.host.protocolID+lightID, s.handleLightStream)
 
 	// register block announce protocol
@@ -590,7 +586,7 @@ func (s *Service) NetworkState() common.NetworkState {
 	}
 }
 
-func (s *Service) TotalConnectedPeers() []peer.ID {
+func (s *Service) AllConnectedPeers() []peer.ID {
 	return s.host.p2pHost.Network().Peers()
 }
 

dot/network/sync.go

@@ -14,37 +14,8 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
-var (
-	blockRequestTimeout = time.Second * 20
-)
-
-func (s *Service) RequestWarpProof(to peer.ID, request *WarpSyncProofRequestMessage) (warpSyncResponse interface{}, err error) {
-	legacyWarpSyncID := s.host.protocolID + warpSyncID
-
-	s.host.p2pHost.ConnManager().Protect(to, "")
-	defer s.host.p2pHost.ConnManager().Unprotect(to, "")
-
-	ctx, cancel := context.WithTimeout(s.ctx, blockRequestTimeout)
-	defer cancel()
-
-	stream, err := s.host.p2pHost.NewStream(ctx, to, legacyWarpSyncID)
-	if err != nil {
-		return nil, err
-	}
-
-	defer func() {
-		err := stream.Close()
-		if err != nil {
-			logger.Warnf("failed to close stream: %s", err)
-		}
-	}()
-
-	if err = s.host.writeToStream(stream, request); err != nil {
-		return nil, err
-	}
-
-	return s.handleWarpSyncProofResponse(stream)
-}
+var blockRequestTimeout = time.Second * 20
+var ErrReceivedEmptyMessage = errors.New("received empty message")
 
 // DoBlockRequest sends a request to the given peer.
 // If a response is received within a certain time period, it is returned,
@@ -77,34 +48,6 @@ func (s *Service) DoBlockRequest(to peer.ID, req *BlockRequestMessage) (*BlockRe
 	return s.receiveBlockResponse(stream)
 }
 
-func (s *Service) handleWarpSyncProofResponse(stream libp2pnetwork.Stream) (interface{}, error) {
-	s.blockResponseBufMu.Lock()
-	defer s.blockResponseBufMu.Unlock()
-
-	// TODO: should we create another buffer pool for warp proof response buffers?
-	buf := s.blockResponseBuf
-
-	n, err := readStream(stream, &buf, warpSyncMaxResponseSize)
-	if err != nil {
-		return nil, fmt.Errorf("reading warp sync stream: %w", err)
-	}
-
-	if n == 0 {
-		return nil, fmt.Errorf("empty warp sync proof")
-	}
-
-	fmt.Printf("WARP PROOF BYTES ---> %v\n", buf[:n])
-	warpProof := new(WarpSyncProofResponse)
-	err = warpProof.Decode(buf[:n])
-	if err != nil {
-		panic(fmt.Sprintf("failed to decode warp proof: %s", err))
-	}
-	fmt.Printf("WARP PROOF ---> %v\n", warpProof)
-	return nil, nil
-}
-
-var ErrReceivedEmptyMessage = errors.New("received empty message")
-
 func (s *Service) receiveBlockResponse(stream libp2pnetwork.Stream) (*BlockResponseMessage, error) {
 	// allocating a new (large) buffer every time slows down the syncing by a dramatic amount,
 	// as malloc is one of the most CPU intensive tasks.

dot/sync/interfaces.go

@@ -82,7 +82,7 @@ type Network interface {
 	// ReportPeer reports peer based on the peer behaviour.
 	ReportPeer(change peerset.ReputationChange, p peer.ID)
 
-	TotalConnectedPeers() []peer.ID
+	AllConnectedPeers() []peer.ID
 }
 
 // Telemetry is the telemetry client to send telemetry messages.

dot/sync/worker_pool.go

@@ -47,7 +47,7 @@ func newSyncWorkerPool(net Network) *syncWorkerPool {
 const ignorePeerTimeout = 2 * time.Minute
 
 func (s *syncWorkerPool) useConnectedPeers() {
-	connectedPeers := s.network.TotalConnectedPeers()
+	connectedPeers := s.network.AllConnectedPeers()
 
 	s.l.Lock()
 	defer s.l.Unlock()