[WIP] magicsock: infinite ping test

Run this for ~2 minutes to see the connection fail and not recover.

wgengine/magicsock/magicsock.go

@@ -67,6 +67,7 @@ type Conn struct {
 	//	10.0.0.3:3 -> [10.0.0.3:3]
 	addrsMu    sync.Mutex
 	addrsByUDP map[udpAddr]*AddrSet
+	addrsByKey map[key.Public]*AddrSet
 
 	// stunReceiveFunc holds the current STUN packet processing func.
 	// Its Loaded value is always non-nil.
@@ -160,6 +161,7 @@ func Listen(opts Options) (*Conn, error) {
 		epFunc:        opts.endpointsFunc(),
 		logf:          log.Printf,
 		addrsByUDP:    make(map[udpAddr]*AddrSet),
+		addrsByKey:    make(map[key.Public]*AddrSet),
 		derpRecvCh:    make(chan derpReadResult),
 		udpRecvCh:     make(chan udpReadResult),
 	}
@@ -770,6 +772,9 @@ func (c *Conn) ReceiveIPv4(b []byte) (n int, ep conn.Endpoint, addr *net.UDPAddr
 				continue
 			}
 
+			// DERP always wins in tests
+			//time.Sleep(100 * time.Millisecond)
+
 			addr := pAddr.(*net.UDPAddr)
 			addr.IP = addr.IP.To4()
 			select {
@@ -808,9 +813,14 @@ func (c *Conn) ReceiveIPv4(b []byte) (n int, ep conn.Endpoint, addr *net.UDPAddr
 			return 0, nil, nil, err
 		}
 
-		// TODO: look up addrSet from dm.Source public key, if
-		// found (Source might be zero for a short period of
-		// time until DERP servers re-deployed)
+		c.addrsMu.Lock()
+		addrSet = c.addrsByKey[dm.src]
+		c.addrsMu.Unlock()
+
+		if addrSet == nil {
+			key := wgcfg.Key(dm.src)
+			log.Printf("magicsock: DERP packet from unknown key: %s", key.ShortString())
+		}
 
 	case um := <-c.udpRecvCh:
 		if um.err != nil {
@@ -1153,6 +1163,7 @@ func (c *Conn) CreateEndpoint(key [32]byte, addrs string) (conn.Endpoint, error)
 		copy(epAddr.ip.Addr[:], addr.IP.To16())
 		epAddr.port = uint16(addr.Port)
 		c.addrsByUDP[epAddr] = a
+		c.addrsByKey[key] = a
 	}
 	c.addrsMu.Unlock()
 

wgengine/magicsock/magicsock_test.go

@@ -13,6 +13,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -519,4 +520,70 @@ func TestTwoDevicePing(t *testing.T) {
 			t.Error("handshake spray failed to find real route")
 		}
 	})
+
+	//cfgs[0] = *dev1.Config()
+	//cfgs[1] = *dev2.Config()
+
+	// Start collecting endpoint updates and Reconfig'ing using them.
+	var cfgsMu sync.Mutex
+	go func() {
+		for epstrs := range epCh1 {
+			eps := append([]wgcfg.Endpoint{derpEp}, makeEps(epstrs)...)
+			//eps[2].Port = 1111
+
+			old := dev2.Config().Peers[0].Endpoints
+			log.Printf("dev2: replacing %v with %v", old, eps)
+
+			cfgsMu.Lock()
+			cfgs[1].Peers[0].Endpoints = eps
+			if err := dev2.Reconfig(&cfgs[1]); err != nil {
+				log.Printf("dev2.Reconfig: %v", err)
+			}
+			cfgsMu.Unlock()
+		}
+	}()
+	go func() {
+		for epstrs := range epCh2 {
+			eps := append([]wgcfg.Endpoint{derpEp}, makeEps(epstrs)...)
+			//eps[2].Port = 1111
+
+			old := dev1.Config().Peers[0].Endpoints
+			log.Printf("dev1: replacing %v with %v", old, eps)
+
+			cfgsMu.Lock()
+			cfgs[0].Peers[0].Endpoints = eps
+			if err := dev1.Reconfig(&cfgs[0]); err != nil {
+				log.Printf("dev1.Reconfig: %v", err)
+			}
+			cfgsMu.Unlock()
+		}
+	}()
+
+	t.Run("infping", func(t *testing.T) {
+		//t.Skipf("used for manual testing only")
+		for i := 0; true; i++ {
+			t.Logf("long running ping %v", time.Now())
+			ping1(t)
+			ping2(t)
+			time.Sleep(1 * time.Second)
+		}
+	})
+}
+
+func makeEps(epstrs []string) (eps []wgcfg.Endpoint) {
+	for _, str := range epstrs {
+		host, port, err := net.SplitHostPort(str)
+		if err != nil {
+			log.Fatalf("bad endpoint %q: %v", str, err)
+		}
+		portNum, err := strconv.Atoi(port)
+		if err != nil {
+			log.Fatalf("bad endpoint port %q: %v", str, err)
+		}
+		eps = append(eps, wgcfg.Endpoint{
+			Host: host,
+			Port: uint16(portNum),
+		})
+	}
+	return eps
 }