Merge pull request #6106 from mysteriumnetwork/add-quic

Fix proper connection handling on multi-consumer setup

Author: soffokl

cmd/commands/connection/command.go

@@ -49,6 +49,12 @@ var (
 		Usage: "Proxy port",
 	}
 
+	flagServiceType = cli.StringFlag{
+		Name:  "service-type",
+		Usage: "Service type to connect to.",
+		Value: serviceWireguard,
+	}
+
 	flagCountry = cli.StringFlag{
 		Name:  "country",
 		Usage: "Two letter (ISO 3166-1 alpha-2) country code to filter proposals.",
@@ -114,7 +120,7 @@ func NewCommand() *cli.Command {
 				Name:      "up",
 				ArgsUsage: "[ProviderIdentityAddress]",
 				Usage:     "Create a new connection",
-				Flags:     []cli.Flag{&config.FlagAgreedTermsConditions, &flagCountry, &flagLocationType, &flagSortType, &flagIncludeFailed, &flagProxyPort},
+				Flags:     []cli.Flag{&config.FlagAgreedTermsConditions, &flagCountry, &flagLocationType, &flagSortType, &flagIncludeFailed, &flagProxyPort, &flagServiceType},
 				Action: func(ctx *cli.Context) error {
 					cmd.up(ctx)
 					return nil
@@ -155,7 +161,7 @@ func (c *command) proposals(ctx *cli.Context) {
 		return
 	}
 
-	proposals, err := c.tequilapi.ProposalsByLocationAndService(serviceWireguard, locationType, locationCountry)
+	proposals, err := c.tequilapi.ProposalsByLocationAndService(ctx.String(flagServiceType.Name), locationType, locationCountry)
 	if err != nil {
 		clio.Warn("Failed to fetch proposal list")
 		return
@@ -291,7 +297,7 @@ func (c *command) up(ctx *cli.Context) {
 		IncludeMonitoringFailed: ctx.Bool(flagIncludeFailed.Name),
 	}
 
-	_, err = c.tequilapi.SmartConnectionCreate(id.Address, hermesID, serviceWireguard, filter, connectOptions)
+	_, err = c.tequilapi.SmartConnectionCreate(id.Address, hermesID, ctx.String(flagServiceType.Name), filter, connectOptions)
 	if err != nil {
 		clio.Error("Failed to create a new connection: ", err)
 		return

p2p/channel_quic.go

@@ -19,6 +19,7 @@ package p2p
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"sync"
@@ -112,6 +113,16 @@ func (c *channelQuic) Send(ctx context.Context, topic string, msg *Message) (*Me
 		return nil, err
 	}
 
+	if readMsg.statusCode != statusCodeOK {
+		if readMsg.statusCode == statusCodePublicErr {
+			return nil, fmt.Errorf("public peer error: %s", string(readMsg.data))
+		}
+		if readMsg.statusCode == statusCodeHandlerNotFoundErr {
+			return nil, fmt.Errorf("%s: %w", string(readMsg.data), ErrHandlerNotFound)
+		}
+		return nil, fmt.Errorf("peer error: %w", errors.New(readMsg.msg))
+	}
+
 	return &Message{
 		Data: readMsg.data,
 	}, nil

p2p/dialer.go

@@ -75,8 +75,6 @@ type dialer struct {
 	verifierFactory identity.VerifierFactory
 	ipResolver      ip.Resolver
 	eventBus        eventbus.EventBus
-
-	quicServer *server.QuicServer
 }
 
 // Dial exchanges p2p configuration via broker, performs NAT pinging if needed
@@ -119,15 +117,16 @@ func (m *dialer) Dial(ctx context.Context, consumerID, providerID identity.Ident
 		}
 	}
 
+	var quicServer *server.QuicServer
 	if serviceType == "quic_scraping" {
-		m.quicServer, err = server.NewQuicServer()
+		quicServer, err = server.NewQuicServer()
 		if err != nil {
 			return nil, fmt.Errorf("could not create QUIC server: %w", err)
 		}
 
-		go m.quicServer.Start(ctx)
+		go quicServer.Start(ctx)
 
-		port, err := m.quicServer.WaitForListenPort(ctx)
+		port, err := quicServer.WaitForListenPort(ctx)
 		if err != nil {
 			return nil, fmt.Errorf("could not wait for listen addr: %w", err)
 		}
@@ -155,7 +154,7 @@ func (m *dialer) Dial(ctx context.Context, consumerID, providerID identity.Ident
 
 	dial := m.dialPinger
 	if len(config.peerURL) > 0 {
-		dial = m.dialQUIC
+		dial = m.dialQUIC(quicServer)
 	} else if len(config.peerPorts) == requiredConnCount {
 		dial = m.dialDirect
 	}
@@ -175,7 +174,7 @@ func (m *dialer) Dial(ctx context.Context, consumerID, providerID identity.Ident
 
 	var channel communicationChannel
 	if serviceType == "quic_scraping" {
-		channel = newChannelQuic(conn1, config.peerID, config.compatibility)
+		channel = newChannelQuic(conn1, providerID, config.compatibility)
 	} else {
 		channel, err = newChannel(conn1, config.privateKey, config.peerPubKey, config.compatibility)
 		if err != nil {
@@ -358,8 +357,19 @@ func (m *dialer) dialDirect(ctx context.Context, providerID identity.Identity, c
 	return conn1, conn2, err
 }
 
-func (m *dialer) dialQUIC(ctx context.Context, providerID identity.Identity, config *p2pConnectConfig) (ServiceConn, ServiceConn, error) {
-	return m.quicServer.CommunicationConn(ctx), m.quicServer.TransportConn(ctx), nil
+func (m *dialer) dialQUIC(quicServer *server.QuicServer) func(ctx context.Context, providerID identity.Identity, config *p2pConnectConfig) (ServiceConn, ServiceConn, error) {
+	return func(ctx context.Context, providerID identity.Identity, config *p2pConnectConfig) (ServiceConn, ServiceConn, error) {
+		conn1, err := quicServer.CommunicationConn(ctx)
+		if err != nil {
+			return nil, nil, fmt.Errorf("could not get QUIC communication connection: %w", err)
+		}
+		conn2, err := quicServer.TransportConn(ctx)
+		if err != nil {
+			return nil, nil, fmt.Errorf("could not get QUIC transport connection: %w", err)
+		}
+
+		return conn1, conn2, nil
+	}
 }
 
 func (m *dialer) dialPinger(ctx context.Context, providerID identity.Identity, config *p2pConnectConfig) (ServiceConn, ServiceConn, error) {

services/quic/connection/server/server.go

@@ -75,34 +75,34 @@ func (s *QuicServer) Start(ctx context.Context) error {
 }
 
 // CommunicationConn returns communication connection.
-func (s *QuicServer) CommunicationConn(ctx context.Context) *streams.QuicConnection {
+func (s *QuicServer) CommunicationConn(ctx context.Context) (*streams.QuicConnection, error) {
 	for {
 		select {
 		case <-ctx.Done():
-			return nil
+			return nil, fmt.Errorf("context done: %w", ctx.Err())
 		default:
 			if s.communicationConn != nil {
-				return &streams.QuicConnection{Connection: s.communicationConn}
+				return &streams.QuicConnection{Connection: s.communicationConn}, nil
 			}
 
-			log.Info().Msg("Waiting for communication connection")
+			log.Debug().Msg("Waiting for communication connection")
 			time.Sleep(200 * time.Millisecond)
 		}
 	}
 }
 
 // TransportConn returns transport connection.
-func (s *QuicServer) TransportConn(ctx context.Context) *streams.QuicConnection {
+func (s *QuicServer) TransportConn(ctx context.Context) (*streams.QuicConnection, error) {
 	for {
 		select {
 		case <-ctx.Done():
-			return nil
+			return nil, fmt.Errorf("context done: %w", ctx.Err())
 		default:
 			if s.transportConn != nil {
-				return &streams.QuicConnection{Connection: s.transportConn}
+				return &streams.QuicConnection{Connection: s.transportConn}, nil
 			}
 
-			log.Info().Msg("Waiting for transport connection")
+			log.Debug().Msg("Waiting for transport connection")
 			time.Sleep(200 * time.Millisecond)
 		}
 	}