connect: proxy fixes

Author: xcolwell

connect/connect_proxy_test.go

@@ -1,3 +1,7 @@
+//go:build !race
+
+// note race detection on this test progressively slows down the test until it stops working
+//
 package main
 
 import (
@@ -11,6 +15,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os/exec"
+	"sync"
 	"time"
 
 	// "strings"
@@ -41,9 +47,24 @@ func testConnectProxy(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	func() {
+		fmt.Printf(`
+
+** This test requires many file descriptions.
+** Setting "ulimit -n 1048576".
+** The value will not be reset when the test ends.
+
+`)
+		_, err := exec.Command("/bin/sh", "-c", "ulimit -n 1048576").CombinedOutput()
+		if err != nil {
+			panic(err)
+		}
+	}()
+
 	maxMessageContentSize := model.ByteCount(4096)
 
-	// sequenceIdleTimeout := 100 * time.Millisecond
+	// clean up lingering residents at this timeout
+	// idleTimeout := 5 * time.Minute
 	// receiveTimeout := 900 * time.Second
 	// minResendInterval := 10 * time.Millisecond
 	// standardContractTransferByteCount := 4 * maxMessageContentSize
@@ -123,8 +144,8 @@ func testConnectProxy(t *testing.T) {
 
 		settings := DefaultExchangeSettings()
 		settings.ExchangeBufferSize = 0
-		// settings.ResidentIdleTimeout = sequenceIdleTimeout
-		// settings.ForwardIdleTimeout = sequenceIdleTimeout
+		// settings.ResidentIdleTimeout = idleTimeout
+		// settings.ForwardIdleTimeout = idleTimeout
 		settings.FramerSettings.MaxMessageLen = int(2 * maxMessageContentSize)
 		settings.ForwardEnforceActiveContracts = true
 		settings.IngressSecurityPolicyGenerator = connect.DisableSecurityPolicyWithStats
@@ -244,7 +265,6 @@ func testConnectProxy(t *testing.T) {
 
 		platformUrl, port := randServerUrl()
 		providerTransportSettings := connect.DefaultPlatformTransportSettings()
-		providerTransportSettings.QuicTlsConfig.InsecureSkipVerify = true
 		providerTransportSettings.H3Port = port + 443
 		providerTransportSettings.DnsPort = port + 53
 		providerTransportSettings.FramerSettings.MaxMessageLen = int(2 * maxMessageContentSize)
@@ -273,7 +293,7 @@ func testConnectProxy(t *testing.T) {
 	}
 
 	providerClientIds := []server.Id{}
-	for range 2 {
+	for range 32 {
 		providerClientId := runProvider()
 		// providerClientId := server.NewId()
 		providerClientIds = append(providerClientIds, providerClientId)
@@ -283,7 +303,7 @@ func testConnectProxy(t *testing.T) {
 	case <-time.After(2 * time.Second):
 	}
 
-	for range 2 {
+	for i := range 1024 {
 		// create a proxy client, connect to a random server, and route to a random provider
 
 		providerClientId := providerClientIds[mathrand.Intn(len(providerClientIds))]
@@ -353,10 +373,10 @@ func testConnectProxy(t *testing.T) {
 		// TODO for some reason go does not pass the host header in the proxy
 		// TODO force the auth token
 		// FIXME see ProxyConnectHeader
-		addr, _ := randHttpsPoxyServerHostPort()
-		httpsProxyUrlStr := fmt.Sprintf("https://%s", addr)
-		httpsProxyUrl, err := url.Parse(httpsProxyUrlStr)
-		assert.Equal(t, err, nil)
+		// addr, _ := randHttpsPoxyServerHostPort()
+		// httpsProxyUrlStr := fmt.Sprintf("https://%s", addr)
+		// httpsProxyUrl, err := url.Parse(httpsProxyUrlStr)
+		// assert.Equal(t, err, nil)
 
 		// proxyDialer := &net.Dialer{
 		// 	Timeout: 15 * time.Second,
@@ -369,26 +389,34 @@ func testConnectProxy(t *testing.T) {
 		proxyConnectHeader.Add("Proxy-Authorization", fmt.Sprintf("Bearer %s", result.ProxyConfigResult.AuthToken))
 
 		proxyHttpClient := &http.Client{
-			Timeout: 60 * time.Second,
+			Timeout: 120 * time.Second,
 			Transport: &http.Transport{
+				// MaxIdleConns:       1,             // Max idle connections in the pool
+				// IdleConnTimeout:    30 * time.Second, // Max time an idle connection stays open
+				// DisableKeepAlives:  true,
 				// TLSHandshakeTimeout: 15 * time.Second,
-				Proxy: http.ProxyURL(httpsProxyUrl),
+				// Proxy: http.ProxyURL(httpsProxyUrl),
+				Proxy: func(r *http.Request) (*url.URL, error) {
+					addr, _ := randHttpsPoxyServerHostPort()
+					httpsProxyUrlStr := fmt.Sprintf("https://%s", addr)
+					return url.Parse(httpsProxyUrlStr)
+				},
 				// DialContext: func(ctx context.Context, network string, addr string) (net.Conn, error) {
 				// 	addr, _ = randServerHostPort()
 				// 	fmt.Printf("DIAL %s VIA %s %s\n", httpsProxyUrl, network, addr)
 				//     return proxyDialer.DialContext(ctx, network, addr)
 				// },
 				ProxyConnectHeader: proxyConnectHeader,
-				TLSClientConfig: &tls.Config{
-					InsecureSkipVerify: true,
-				},
 			},
 		}
 
-		for range 2 {
+		runOne := func() {
+			// var err error
+			// for range 2 {
 			addr, _ := randServerHostPort()
 			targetUrlString := fmt.Sprintf("https://%s/status", addr)
 			// targetUrlString := "https://api.bringyour.com/hello"
+			// var r *http.Response
 			r, err := proxyHttpClient.Get(targetUrlString)
 			assert.Equal(t, err, nil)
 			if err == nil {
@@ -399,8 +427,33 @@ func testConnectProxy(t *testing.T) {
 				err = json.Unmarshal(b, &m)
 				assert.Equal(t, err, nil)
 				assert.NotEqual(t, m["client_address"], nil)
+				return
 			}
+			// FIXME
+			// else something flaky about the proxy connection
+			// try again
+			// }
+			// assert.Equal(t, err, nil)
+		}
+
+		// run in serial
+		for j := range 32 {
+			fmt.Printf("[%d][%d][s]start\n", i, j)
+			runOne()
+		}
+
+		// run in parallel
+		var wg sync.WaitGroup
+		for j := range 32 {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				fmt.Printf("[%d][%d][p]start\n", i, j)
+				runOne()
+			}()
 		}
+		wg.Wait()
 
 		proxyHttpClient.CloseIdleConnections()
 

connect/resident.go

@@ -1491,6 +1491,8 @@ func NewResident(
 	instanceId server.Id,
 	residentId server.Id,
 ) *Resident {
+	glog.V(1).Infof("[r]create")
+
 	cancelCtx, cancel := context.WithCancel(ctx)
 
 	proxyDeviceConfig := model.GetProxyDeviceConfigForClient(cancelCtx, clientId, instanceId)
@@ -1682,7 +1684,9 @@ func (self *Resident) handleClientForward(path connect.TransferPath, transferFra
 
 	// FIXME deep packet inspection to look at the contract frames and verify contracts before forwarding
 
-	c := func() bool {
+	initForward := func() *ResidentForward {
+		self.stateLock.Lock()
+		defer self.stateLock.Unlock()
 
 		nextForward := func() *ResidentForward {
 			if self.exchange.settings.ForwardEnforceActiveContracts {
@@ -1729,43 +1733,50 @@ func (self *Resident) handleClientForward(path connect.TransferPath, transferFra
 				}
 			})
 
-			var replacedForward *ResidentForward
-			func() {
-				self.stateLock.Lock()
-				defer self.stateLock.Unlock()
-				replacedForward = self.forwards[destinationId]
-				self.forwards[destinationId] = forward
-			}()
-			if replacedForward != nil {
-				replacedForward.Cancel()
-			}
-			glog.V(1).Infof("[rf]open %s->%s\n", sourceId, destinationId)
-
 			return forward
 		}
 
 		limit := false
 		var forward *ResidentForward
-		func() {
-			self.stateLock.Lock()
-			defer self.stateLock.Unlock()
-			var ok bool
-			forward, ok = self.forwards[destinationId]
-			if !ok && self.exchange.settings.MaxConcurrentForwardsPerResident <= len(self.forwards) {
-				limit = true
-			}
-		}()
+		// func() {
+		// 	self.stateLock.Lock()
+		// 	defer self.stateLock.Unlock()
+		var ok bool
+		forward, ok = self.forwards[destinationId]
+		if !ok && self.exchange.settings.MaxConcurrentForwardsPerResident <= len(self.forwards) {
+			limit = true
+		}
+		// }()
 
 		if forward == nil && limit {
 			glog.Infof("[rf]abuse forward limit %s->%s", sourceId, destinationId)
 			self.abuseLimiter.delay()
-			return false
+			return nil
 		}
 
 		if forward == nil || !forward.UpdateActivity() {
 			forward = nextForward()
+
+			var replacedForward *ResidentForward
+			// func() {
+			// 	self.stateLock.Lock()
+			// 	defer self.stateLock.Unlock()
+			replacedForward = self.forwards[destinationId]
+			self.forwards[destinationId] = forward
+			// }()
+			if replacedForward != nil {
+				replacedForward.Cancel()
+			}
+			glog.V(1).Infof("[rf]open %s->%s\n", sourceId, destinationId)
+
 		}
 
+		return forward
+	}
+
+	c := func() bool {
+		forward := initForward()
+
 		if forward == nil {
 			return false
 		}

connect/resident_proxy.go

@@ -7,6 +7,8 @@ import (
 	// "fmt"
 	"time"
 
+	"github.com/urnetwork/glog"
+
 	"github.com/urnetwork/connect"
 	"github.com/urnetwork/connect/protocol"
 	"github.com/urnetwork/sdk"
@@ -67,6 +69,7 @@ func NewResidentProxyDevice(
 	proxyDeviceConfig *model.ProxyDeviceConfig,
 	settings *ResidentProxyDeviceSettings,
 ) (*ResidentProxyDevice, error) {
+	glog.Infof("[rp]create")
 
 	// this jwt is used to access the services in the network space
 	byJwt, err := jwt.LoadByJwtFromClientId(ctx, clientId)
@@ -133,6 +136,8 @@ func (self *ResidentProxyDevice) AddTun() (
 	receive chan []byte,
 	closeTun func(),
 ) {
+	glog.Infof("[rp]add tun")
+
 	send = make(chan []byte, self.exchange.settings.ExchangeBufferSize)
 	receive = make(chan []byte, self.exchange.settings.ExchangeBufferSize)
 

connect/resident_proxy_generator.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/urnetwork/connect"
 	"github.com/urnetwork/connect/protocol"
+	"github.com/urnetwork/glog"
 	"github.com/urnetwork/server"
 	"github.com/urnetwork/server/controller"
 	"github.com/urnetwork/server/jwt"
@@ -72,6 +73,10 @@ func (self *exchangeGenerator) clientSession() *session.ClientSession {
 	return session.NewLocalClientSession(self.ctx, "127.0.0.1", self.byJwt)
 }
 
+func (self *exchangeGenerator) cleanupSession() *session.ClientSession {
+	return session.NewLocalClientSession(context.Background(), "127.0.0.1", self.byJwt)
+}
+
 func (self *exchangeGenerator) NextDestinations(count int, excludeDestinations []connect.MultiHopId, rankMode string) (map[connect.MultiHopId]connect.DestinationStats, error) {
 	excludeDestinationsIds := [][]server.Id{}
 	for _, excludeDestination := range excludeDestinations {
@@ -169,7 +174,7 @@ func (self *exchangeGenerator) RemoveClientArgs(args *connect.MultiClientGenerat
 		ClientId: server.Id(args.ClientId),
 	}
 
-	model.RemoveNetworkClient(removeNetworkClient, self.clientSession())
+	model.RemoveNetworkClient(removeNetworkClient, self.cleanupSession())
 }
 
 func (self *exchangeGenerator) RemoveClientWithArgs(client *connect.Client, args *connect.MultiClientGeneratorClientArgs) {