p2p: speed up startup and initial peer connection

Author: pymq

application.go

@@ -91,6 +91,8 @@ func New() *Application {
 }
 
 func (a *Application) Init(ctx context.Context, tunDevice tun.Device) error {
+	a.logger.Info("Application initialization started")
+
 	a.ctx, a.ctxCancel = context.WithCancel(ctx)
 	a.P2p = p2p.NewP2p(a.ctx)
 	p2pHost, err := a.P2p.InitHost(a.makeP2pHostConfig())
@@ -100,8 +102,8 @@ func (a *Application) Init(ctx context.Context, tunDevice tun.Device) error {
 
 	privKey := p2pHost.Peerstore().PrivKey(p2pHost.ID())
 	a.Conf.SetIdentity(privKey, p2pHost.ID())
-	a.logger.Infof("Host created. We are: %s", p2pHost.ID().String())
-	a.logger.Infof("Listen interfaces: %v", p2pHost.Addrs())
+	a.logger.Infof("P2P host initialized. My peer_id: %s", p2pHost.ID().String())
+	a.logger.Infof("P2P listening on addresses: %v", p2pHost.Addrs())
 
 	localIP, netMask := a.Conf.VPNLocalIPMask()
 	interfaceName := a.Conf.VPNConfig.InterfaceName
@@ -110,12 +112,9 @@ func (a *Application) Init(ctx context.Context, tunDevice tun.Device) error {
 		return fmt.Errorf("failed to init vpn: %v", err)
 	}
 	a.vpnDevice = vpnDevice
-	a.logger.Infof("Created vpn interface %s: %s", interfaceName, &net.IPNet{IP: localIP, Mask: netMask})
+	a.logger.Infof("VPN interface created. Name: %s CIDR: %s", interfaceName, &net.IPNet{IP: localIP, Mask: netMask})
 
-	err = a.P2p.Bootstrap()
-	if err != nil {
-		return err
-	}
+	a.P2p.Bootstrap()
 
 	a.Dns = NewDNSService(a.Conf, a.Eventbus, a.ctx, a.logger)
 	a.AuthStatus = service.NewAuthStatus(a.P2p, a.Conf, a.Eventbus)
@@ -151,11 +150,13 @@ func (a *Application) Init(ctx context.Context, tunDevice tun.Device) error {
 		interfaceName, err := a.vpnDevice.InterfaceName()
 		if err != nil {
 			a.logger.Errorf("failed to get TUN interface name: %v", err)
-			return nil
+		} else {
+			a.Dns.initDNS(interfaceName)
 		}
-		a.Dns.initDNS(interfaceName)
 	}
 
+	a.logger.Info("Application initialized successfully")
+
 	return nil
 }
 
@@ -176,7 +177,7 @@ func (a *Application) SetupLoggerAndConfig() *log.ZapEventLogger {
 
 	encoderConfig := zap.NewDevelopmentEncoderConfig()
 	encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
-		enc.AppendString(t.Format("2006-01-02 15:04:05"))
+		enc.AppendString(t.Format("2006-01-02 15:04:05.99"))
 	}
 	consoleEncoder := zapcore.NewConsoleEncoder(encoderConfig)
 	zapCore := zapcore.NewCore(consoleEncoder, syncer, zapcore.InfoLevel)

p2p/p2p.go

@@ -95,6 +95,8 @@ type P2p struct {
 	bootstrapPeers   []peer.AddrInfo
 	startedAt        time.Time
 	bootstrapsInfo   atomic.Pointer[map[string]BootstrapPeerDebugInfo]
+
+	dhtBootstrapFinishedChan chan struct{}
 }
 
 func NewP2p(ctx context.Context) *P2p {
@@ -103,6 +105,8 @@ func NewP2p(ctx context.Context) *P2p {
 		ctx:       newCtx,
 		ctxCancel: ctxCancel,
 		logger:    log.Logger("awl/p2p"),
+
+		dhtBootstrapFinishedChan: make(chan struct{}),
 	}
 }
 
@@ -300,45 +304,68 @@ func (p *P2p) SubscribeConnectionEvents(onConnected, onDisconnected func(network
 	p.host.Network().Notify(notifyBundle)
 }
 
-func (p *P2p) Bootstrap() error {
-	p.logger.Debug("Bootstrapping the DHT")
-	// connect to the bootstrap nodes first
-	ctx, cancel := context.WithTimeout(p.ctx, 2*time.Second)
-	defer cancel()
+func (p *P2p) Bootstrap() {
+	ctx, cancel := context.WithTimeout(p.ctx, 3*time.Second)
 	var wg sync.WaitGroup
+	successfulConnectionsCh := make(chan struct{}, len(p.bootstrapPeers))
 
+	p.logger.Debug("Start bootstrapping the DHT")
 	for _, peerAddr := range p.bootstrapPeers {
 		wg.Add(1)
-		p.host.ConnManager().Protect(peerAddr.ID, protectedBootstrapPeerTag)
-
 		go func() {
 			defer wg.Done()
 			if err := p.host.Connect(ctx, peerAddr); err != nil && !errors.Is(err, context.Canceled) {
 				p.logger.Warnf("Failed to connect to bootstrap node %s: %v", peerAddr.ID, err)
 			} else if err == nil {
 				p.logger.Infof("Connection established with bootstrap node: %s", peerAddr.ID)
+				successfulConnectionsCh <- struct{}{}
 			}
 		}()
 	}
-	wg.Wait()
-	p.logger.Info("Connection established with all bootstrap nodes")
 
-	if err := p.dht.Bootstrap(p.ctx); err != nil {
-		return fmt.Errorf("bootstrap dht: %v", err)
-	}
+	go func() {
+		defer cancel()
 
-	return nil
+		// wait for at least 2 bootstrap nodes
+		for range min(2, len(p.bootstrapPeers)) {
+			select {
+			case <-ctx.Done():
+			case <-successfulConnectionsCh:
+			}
+		}
+
+		p.logger.Info("Bootstrapping the DHT")
+		if err := p.dht.Bootstrap(p.ctx); err != nil {
+			// from the code err is always nil for now
+			p.logger.Warnf("Failed to bootstrap DHT: %v", err)
+		}
+		close(p.dhtBootstrapFinishedChan)
+
+		wg.Wait()
+		p.logger.Info("Finished connecting to all bootstrap nodes")
+	}()
 }
 
 func (p *P2p) MaintainBackgroundConnections(ctx context.Context, interval time.Duration, knownPeersIdsFunc func() []peer.ID) {
-	const firstTryInterval = 5 * time.Second
-	p.connectToKnownPeers(ctx, firstTryInterval, knownPeersIdsFunc())
+	const timeout = 5 * time.Second
+	const firstRetryDelay = 5 * time.Second
+
+	// wait for bootstrapping
+	select {
+	case <-ctx.Done():
+		return
+	case <-p.dhtBootstrapFinishedChan:
+	}
+
+	p.connectToKnownPeers(ctx, timeout, knownPeersIdsFunc())
+
+	// retry once after a short delay in case of network instability
 	select {
 	case <-ctx.Done():
 		return
-	case <-time.After(firstTryInterval):
+	case <-time.After(firstRetryDelay):
 	}
-	p.connectToKnownPeers(ctx, interval, knownPeersIdsFunc())
+	p.connectToKnownPeers(ctx, timeout, knownPeersIdsFunc())
 
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -350,7 +377,7 @@ func (p *P2p) MaintainBackgroundConnections(ctx context.Context, interval time.D
 		case <-ticker.C:
 		}
 
-		p.connectToKnownPeers(ctx, interval, knownPeersIdsFunc())
+		p.connectToKnownPeers(ctx, timeout, knownPeersIdsFunc())
 		ticker.Reset(interval)
 	}
 }
@@ -380,6 +407,8 @@ func (p *P2p) connectToKnownPeers(ctx context.Context, timeout time.Duration, pe
 
 	for _, peerAddr := range p.bootstrapPeers {
 		wg.Add(1)
+		p.host.ConnManager().Protect(peerAddr.ID, protectedBootstrapPeerTag)
+
 		go func() {
 			defer wg.Done()
 

service/auth_status.go

@@ -30,7 +30,6 @@ type P2p interface {
 	NewStream(ctx context.Context, id peer.ID, proto libp2pProtocol.ID) (network.Stream, error)
 	NewStreamWithDedicatedConn(ctx context.Context, id peer.ID, proto libp2pProtocol.ID) (network.Stream, error)
 	SubscribeConnectionEvents(onConnected, onDisconnected func(network.Network, network.Conn))
-	ProtectPeer(id peer.ID)
 	RecordPeerLatency(id peer.ID, rtt time.Duration)
 }
 
@@ -338,7 +337,6 @@ func (s *AuthStatus) AddPeer(ctx context.Context, peerID peer.ID, name, alias st
 	newPeerConfig.DomainName = awldns.TrimDomainName(newPeerConfig.DisplayName())
 	s.conf.RemoveBlockedPeer(peerIDStr)
 	s.conf.UpsertPeer(newPeerConfig)
-	s.p2p.ProtectPeer(peerID)
 
 	go func() {
 		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)

test_suite_test.go

@@ -204,8 +204,7 @@ func (ts *TestSuite) initBootstrapNode() {
 	p2pSrv := p2p.NewP2p(context.Background())
 	p2pHost, err := p2pSrv.InitHost(hostConfig)
 	ts.NoError(err)
-	err = p2pSrv.Bootstrap()
-	ts.NoError(err)
+	p2pSrv.Bootstrap()
 
 	peerInfo := peer.AddrInfo{ID: p2pHost.ID(), Addrs: p2pHost.Addrs()}
 	ts.bootstrapAddrs = append(ts.bootstrapAddrs, peerInfo)
@@ -221,15 +220,12 @@ func (ts *TestSuite) initBootstrapNode() {
 }
 
 func (ts *TestSuite) ensurePeersAvailableInDHT(peer1, peer2 TestPeer) {
-	ts.Eventually(func() bool {
-		err1 := peer1.app.P2p.Bootstrap()
-		err2 := peer2.app.P2p.Bootstrap()
-		if err1 != nil || err2 != nil {
-			return false
-		}
+	peer1.app.P2p.Bootstrap()
+	peer2.app.P2p.Bootstrap()
 
-		_, err1 = peer1.app.P2p.FindPeer(context.Background(), peer2.app.P2p.PeerID())
-		_, err2 = peer2.app.P2p.FindPeer(context.Background(), peer1.app.P2p.PeerID())
+	ts.Eventually(func() bool {
+		_, err1 := peer1.app.P2p.FindPeer(context.Background(), peer2.app.P2p.PeerID())
+		_, err2 := peer2.app.P2p.FindPeer(context.Background(), peer1.app.P2p.PeerID())
 
 		return err1 == nil && err2 == nil
 	}, 20*time.Second, 100*time.Millisecond)