feat:dot/network: open block announce stream and send handshake on connect

dot/network/service.go

@@ -178,7 +178,6 @@ func (s *Service) Start() error {
 		s.syncQueue.peerScore.Delete(p)
 	})
 
-	s.host.h.Network().SetConnHandler(s.handleConn)
 	s.host.registerStreamHandler(syncID, s.handleSyncStream)
 	s.host.registerStreamHandler(lightID, s.handleLightStream)
 
@@ -212,6 +211,9 @@ func (s *Service) Start() error {
 		logger.Warn("failed to register notifications protocol", "sub-protocol", blockAnnounceID, "error", err)
 	}
 
+	// since this opens block announce streams, it should happen after the protocol is registered
+	s.host.h.Network().SetConnHandler(s.handleConn)
+
 	// log listening addresses to console
 	for _, addr := range s.host.multiaddrs() {
 		logger.Info("Started listening", "address", addr)
@@ -280,6 +282,39 @@ func (s *Service) logPeerCount() {
 func (s *Service) handleConn(conn libp2pnetwork.Conn) {
 	// give new peers a slight weight
 	s.syncQueue.updatePeerScore(conn.RemotePeer(), 1)
+
+	s.notificationsMu.Lock()
+	defer s.notificationsMu.Unlock()
+
+	info, has := s.notificationsProtocols[BlockAnnounceMsgType]
+	if !has {
+		// this shouldn't happen
+		logger.Warn("block announce protocol is not yet registered!")
+		return
+	}
+
+	// open block announce substream
+	hs, err := info.getHandshake()
+	if err != nil {
+		logger.Warn("failed to get handshake", "protocol", blockAnnounceID, "error", err)
+		return
+	}
+
+	info.mapMu.RLock()
+	defer info.mapMu.RUnlock()
+
+	peer := conn.RemotePeer()
+	if hsData, has := info.handshakeData[peer]; !has || !hsData.received {
+		info.handshakeData[peer] = &handshakeData{
+			validated: false,
+		}
+
+		logger.Trace("sending handshake", "protocol", info.protocolID, "peer", peer, "message", hs)
+		err = s.host.send(peer, info.protocolID, hs)
+		if err != nil {
+			logger.Trace("failed to send block announce handshake to peer", "peer", peer, "error", err)
+		}
+	}
 }
 
 func (s *Service) beginDiscovery() error {

dot/network/service_test.go

@@ -329,3 +329,56 @@ func TestPersistPeerStore(t *testing.T) {
 	nodeAA := createTestService(t, nodeA.cfg)
 	require.NotEmpty(t, nodeAA.host.h.Peerstore().PeerInfo(nodeB.host.id()).Addrs)
 }
+
+func TestHandleConn(t *testing.T) {
+	configA := &Config{
+		BasePath:    utils.NewTestBasePath(t, "nodeA"),
+		Port:        7001,
+		RandSeed:    1,
+		NoBootstrap: true,
+		NoMDNS:      true,
+	}
+
+	nodeA := createTestService(t, configA)
+
+	configB := &Config{
+		BasePath:    utils.NewTestBasePath(t, "nodeB"),
+		Port:        7002,
+		RandSeed:    2,
+		NoBootstrap: true,
+		NoMDNS:      true,
+	}
+
+	nodeB := createTestService(t, configB)
+
+	addrInfosB, err := nodeB.host.addrInfos()
+	require.NoError(t, err)
+
+	err = nodeA.host.connect(*addrInfosB[0])
+	if failedToDial(err) {
+		time.Sleep(TestBackoffTimeout)
+		err = nodeA.host.connect(*addrInfosB[0])
+	}
+	require.NoError(t, err)
+
+	time.Sleep(time.Second)
+
+	bScore, ok := nodeA.syncQueue.peerScore.Load(nodeB.host.id())
+	require.True(t, ok)
+	require.Equal(t, 1, bScore)
+	aScore, ok := nodeB.syncQueue.peerScore.Load(nodeA.host.id())
+	require.True(t, ok)
+	require.Equal(t, 1, aScore)
+
+	infoA := nodeA.notificationsProtocols[BlockAnnounceMsgType]
+	hsDataB, has := infoA.handshakeData[nodeB.host.id()]
+	require.True(t, has)
+	require.True(t, hsDataB.received)
+	require.True(t, hsDataB.validated)
+
+	infoB := nodeB.notificationsProtocols[BlockAnnounceMsgType]
+	hsDataA, has := infoB.handshakeData[nodeA.host.id()]
+	require.True(t, has)
+	require.True(t, hsDataA.received)
+	require.True(t, hsDataA.validated)
+}