Find a refused packet's tunnel from its header, ask the lighthouse when probing, and address review

connection_manager_test.go

@@ -8,9 +8,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/flynn/noise"
 	"github.com/rcrowley/go-metrics"
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/header"
+	"github.com/slackhq/nebula/noiseutil"
+	"github.com/slackhq/nebula/overlay/batch"
 	"github.com/slackhq/nebula/overlay/overlaytest"
 	"github.com/slackhq/nebula/test"
 	"github.com/slackhq/nebula/udp"
@@ -607,6 +611,9 @@ func newPunchTestConnectionManager(t *testing.T, conn udp.Conn) (*connectionMana
 	nc := newConnectionManagerFromConfig(l, config.NewC(l), hostMap, punchy)
 	nc.intf = ifce
 	ifce.connectionManager = nc
+	// Tunnel sends go through conn too, though without a key from withTestKey they never get as far as writing
+	ifce.writers = []udp.Conn{conn}
+	ifce.metricTxDropped = metrics.NilCounter{}
 
 	hostinfo := newPunchTarget()
 	hostinfo.localIndexId = 1099
@@ -657,6 +664,13 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) {
 		nc.sendPunch(hostinfo)
 		assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote)
 		assert.NotZero(t, hostinfo.nextUnreachableProbe.Load(), "the tunnel should have probed its other remotes")
+		assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse),
+			"the probe should ask the lighthouse to have the peer punch us")
+
+		// Within the probe interval nothing more goes out, however many sends fail
+		nc.sendPunch(hostinfo)
+		nc.sendPunch(hostinfo)
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
 	})
 
 	t.Run("delivered", func(t *testing.T) {
@@ -666,6 +680,7 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) {
 		nc.sendPunch(hostinfo)
 		assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote)
 		assert.Zero(t, hostinfo.nextUnreachableProbe.Load())
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
 	})
 
 	t.Run("other failure", func(t *testing.T) {
@@ -674,5 +689,133 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) {
 
 		nc.sendPunch(hostinfo)
 		assert.Zero(t, hostinfo.nextUnreachableProbe.Load(), "only a missing route says to look elsewhere")
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
+	})
+}
+
+// A packet the data path had no route for names its tunnel in its header, the peer's index for it
+func Test_sendUnreachable_findsTunnelFromHeader(t *testing.T) {
+	head := func(st header.MessageSubType, remoteIndex uint32) []byte {
+		return header.Encode(make([]byte, header.Len), header.Version, header.Message, st, remoteIndex, 1)
+	}
+
+	t.Run("direct", func(t *testing.T) {
+		nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{})
+		nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId), punchTargetV6)
+		assert.NotZero(t, hostinfo.nextUnreachableProbe.Load(), "the tunnel should have probed its other remotes")
+		assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse))
+	})
+
+	t.Run("not its remote", func(t *testing.T) {
+		nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{})
+		nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId), punchTargetV4)
+		assert.Zero(t, hostinfo.nextUnreachableProbe.Load(), "a failure on a remote the tunnel is not using is stale")
+	})
+
+	t.Run("relayed", func(t *testing.T) {
+		nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{})
+		// A relay header's index is the relay session's, it must not be taken for a tunnel's even when they collide
+		nc.intf.sendUnreachable(head(header.MessageRelay, hostinfo.remoteIndexId), punchTargetV6)
+		assert.Zero(t, hostinfo.nextUnreachableProbe.Load())
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
+	})
+
+	t.Run("unknown tunnel", func(t *testing.T) {
+		nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{})
+		nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId+1), punchTargetV6)
+		assert.Zero(t, hostinfo.nextUnreachableProbe.Load())
+	})
+}
+
+// withTestKey gives a tunnel a real key so its sends get encrypted and written instead of returning early
+func withTestKey(hostinfo *HostInfo) {
+	cs := noise.NewCipherSuite(noise.DH25519, noise.CipherAESGCM, noise.HashSHA256)
+	hostinfo.ConnectionState.eKey = noiseutil.NewCipherState(noise.UnsafeNewCipherState(cs, [32]byte{}, 0), noise.CipherAESGCM)
+}
+
+// The probe itself, end to end from a failed send: a test packet to every remote but the one we cannot reach, and a
+// lighthouse query, through both places a send can fail.
+func Test_remoteUnreachable_probesOtherRemotes(t *testing.T) {
+	t.Run("control send", func(t *testing.T) {
+		conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}}
+		nc, hostinfo := newPunchTestConnectionManager(t, conn)
+		withTestKey(hostinfo)
+
+		nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu))
+		assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV4}, conn.wrote, "the failed send, then the probe")
+		assert.Equal(t, []header.MessageType{header.Test, header.Test}, conn.types)
+		assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse))
+	})
+
+	t.Run("data path batch", func(t *testing.T) {
+		conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}}
+		nc, hostinfo := newPunchTestConnectionManager(t, conn)
+		withTestKey(hostinfo)
+
+		// Two of the tunnel's packets fail in one batch, they probe once
+		sb := batch.NewSendBatch(conn, 4, 4096)
+		for c := uint64(1); c <= 2; c++ {
+			pkt := header.Encode(sb.Reserve(64), header.Version, header.Message, header.MessageNone, hostinfo.remoteIndexId, c)
+			sb.Commit(append(pkt, make([]byte, 32)...), punchTargetV6)
+		}
+		nc.intf.flushSendBatch(sb, 0)
+
+		assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV6, punchTargetV4}, conn.wrote,
+			"the failed packets, then one probe")
+		assert.Equal(t, header.Test, conn.types[2])
+		assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse))
 	})
+
+	t.Run("closing", func(t *testing.T) {
+		conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}}
+		nc, hostinfo := newPunchTestConnectionManager(t, conn)
+		withTestKey(hostinfo)
+
+		nc.intf.SendMessageToHostInfo(header.CloseTunnel, 0, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu))
+		assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote, "a tunnel being torn down does not probe")
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
+	})
+
+	t.Run("lighthouse", func(t *testing.T) {
+		conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}}
+		nc, hostinfo := newPunchTestConnectionManager(t, conn)
+		withTestKey(hostinfo)
+		nc.intf.lightHouse.amLighthouse = true
+
+		nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu))
+		assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote, "a lighthouse does not roam its clients")
+	})
+
+	t.Run("nothing else to probe", func(t *testing.T) {
+		conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}}
+		nc, hostinfo := newPunchTestConnectionManager(t, conn)
+		withTestKey(hostinfo)
+		nc.intf.reQueryWait.Store(int64(time.Minute))
+		// Forget the IPv4 remote, all the tunnel knows is the one it cannot reach
+		hostinfo.remotes = NewRemoteList(hostinfo.vpnAddrs, nil)
+		hostinfo.remotes.LearnRemote(hostinfo.vpnAddrs[0], punchTargetV6)
+
+		nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu))
+		assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse),
+			"the lighthouse is the only way to learn of another remote")
+
+		// Past the probe interval, but the lighthouse is only asked again on the requery schedule
+		hostinfo.nextUnreachableProbe.Store(0)
+		nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu))
+		assert.Empty(t, drainQueries(nc.intf.lightHouse))
+		assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV6}, conn.wrote)
+	})
+}
+
+// drainQueries returns every lighthouse query waiting to go out
+func drainQueries(lh *LightHouse) []netip.Addr {
+	var queries []netip.Addr
+	for {
+		select {
+		case addr := <-lh.queryChan:
+			queries = append(queries, addr)
+		default:
+			return queries
+		}
+	}
 }

control_tester.go

@@ -117,6 +117,11 @@ func (c *Control) SetUDPAddr(addr netip.AddrPort) {
 	c.f.outside.(*udp.TesterConn).SetAddr(addr)
 }
 
+// GetUDPAddrFor returns the underlay address this node sends to dst from, which differs by family for a dual stack node
+func (c *Control) GetUDPAddrFor(dst netip.AddrPort) netip.AddrPort {
+	return c.f.outside.(*udp.TesterConn).GetAddrFor(dst)
+}
+
 // SetDualStackUDPAddrs gives this node one underlay address per family, sending from whichever matches the
 // destination. An invalid AddrPort leaves the node without that family, sends to it fail with ENETUNREACH. Register
 // any address the node did not start on with the router as well or nothing will route back.

e2e/dual_stack_roam_test.go

@@ -33,6 +33,8 @@ func TestDualStackRoamToV4Only(t *testing.T) {
 		{name: "peer sends, lh over v4", themSends: true, lhOverV4: true},
 		{name: "peer sends, punchy", themSends: true, punchy: true},
 		{name: "peer sends, punchy all", themSends: true, punchy: true, punchAll: true},
+		{name: "laptop sends, peer NAT", meSends: true, peerNAT: true},
+		{name: "laptop sends, peer NAT, lh v4", meSends: true, peerNAT: true, lhOverV4: true},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
@@ -42,6 +44,26 @@ func TestDualStackRoamToV4Only(t *testing.T) {
 	}
 }
 
+// TestDualStackRoamRecovers holds the line TestDualStackRoamToV4Only measures: a tunnel whose remote became unreachable
+// moves to the peer's other address instead of waiting out the connection manager to rehandshake, which takes 20s.
+func TestDualStackRoamRecovers(t *testing.T) {
+	for _, tc := range []dualStackRoamCase{
+		{name: "both send", meSends: true, themSends: true},
+		{name: "laptop sends, peer NAT", meSends: true, peerNAT: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			fwd, rev := measureDualStackRoam(t, tc)
+			if fwd == 0 || fwd > 5*time.Second {
+				t.Errorf("laptop->peer recovered in %v, want within 5s", fmtRecovery(fwd))
+			}
+			if tc.themSends && (rev == 0 || rev > 5*time.Second) {
+				t.Errorf("peer->laptop recovered in %v, want within 5s", fmtRecovery(rev))
+			}
+		})
+	}
+}
+
 type dualStackRoamCase struct {
 	name string
 	// meSends and themSends say who keeps traffic flowing after the move
@@ -55,6 +77,8 @@ type dualStackRoamCase struct {
 	simultaneous bool
 	// punchy turns on punchy.punch everywhere, punchAll adds punchy.target_all_remotes
 	punchy, punchAll bool
+	// peerNAT puts the peer behind a NAT that, after the move, only lets in traffic from addresses the peer has sent to
+	peerNAT bool
 }
 
 func fmtRecovery(d time.Duration) string {
@@ -226,10 +250,21 @@ func measureDualStackRoam(t *testing.T, tc dualStackRoamCase) (time.Duration, ti
 	done := func() bool {
 		return (!tc.meSends || fwd.Load() != 0) && (!tc.themSends || rev.Load() != 0)
 	}
+	// The peer's NAT has state for the lighthouse, which its own updates keep fresh, and for wherever it sends from here
+	natOpen := map[netip.AddrPort]bool{lhV4: true, lhV6: true}
+	isPeer := func(a netip.AddrPort) bool { return a == them.v4 || a == them.v6 }
+
 	r.RouteForAllExitFuncOrTimeout(time.Second*90, func(p *udp.Packet, _ *nebula.Control) router.ExitType {
 		if p.To == officeV4 || p.To == officeV6Addr {
 			return router.Drop
 		}
+		if tc.peerNAT {
+			if isPeer(p.From) {
+				natOpen[p.To] = true
+			} else if isPeer(p.To) && !natOpen[p.From] {
+				return router.Drop
+			}
+		}
 		if done() {
 			return router.ExitNow
 		}

e2e/router/router.go

@@ -967,7 +967,8 @@ func (r *R) getControl(p *udp.Packet) *nebula.Control {
 
 	c, ok := r.inNat[toAddr]
 	if ok {
-		r.outNat[outNatKey{from: c.GetUDPAddr(), to: fromAddr}] = toAddr
+		// Key the return path on the address the receiver will answer from, which is per family for a dual stack node
+		r.outNat[outNatKey{from: c.GetUDPAddrFor(fromAddr), to: fromAddr}] = toAddr
 		return c
 	}
 

hostmap.go

@@ -775,11 +775,18 @@ const unreachableProbeInterval = time.Second
 // remoteUnreachable is called when a send to our current remote failed because we have no route to it, typically
 // after moving to a network without that address family. Sends a test packet to every other remote we know for this
 // host: the first one that arrives roams the peer over to wherever we are now, and its reply roams us onto that path.
+//
+// A peer behind a NAT drops a test from an address it has never sent to, so every probe also asks the lighthouse
+// about the peer. The lighthouse tells the peer where we are now, the peer punches us, and the next probe gets in.
 func (i *HostInfo) remoteUnreachable(ifce *Interface, dead netip.AddrPort) {
 	if i.remotes == nil || i.ConnectionState == nil || i.GetRemote() != dead {
 		// Nothing to probe with, or we already moved on from the remote that failed
 		return
 	}
+	if ifce.lightHouse.amLighthouse {
+		// A lighthouse hands out the address it reaches a client on, it must not go roaming its clients itself
+		return
+	}
 
 	now := time.Now().UnixNano()
 	next := i.nextUnreachableProbe.Load()
@@ -789,12 +796,24 @@ func (i *HostInfo) remoteUnreachable(ifce *Interface, dead netip.AddrPort) {
 
 	nb := make([]byte, 12)
 	out := make([]byte, mtu)
+	probed := 0
 	i.remotes.ForEach(ifce.hostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) {
 		if addr == dead {
 			return
 		}
+		probed++
 		ifce.sendTo(header.Test, header.TestRequest, i.ConnectionState, i, addr, []byte(""), nb, out)
 	})
+
+	if probed == 0 {
+		// With nothing else to try, the lighthouse is our only way to learn of another remote. The peer may not have
+		// one, so ask on the usual requery schedule rather than every probe for as long as the tunnel sits here.
+		if now < i.nextLHQuery.Load() {
+			return
+		}
+		i.nextLHQuery.Store(now + ifce.reQueryWait.Load())
+	}
+	ifce.lightHouse.QueryServer(i.vpnAddrs[0])
 }
 
 // Bits within HostInfo.state, everything above stateEpochShift is the epoch

inside.go

@@ -215,7 +215,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b
 				return nil
 			}
 
-			sendBatch.Commit(toSend, relayHostInfo.GetRemote(), relayHostInfo.localIndexId)
+			sendBatch.Commit(toSend, relayHostInfo.GetRemote())
 			return nil
 		})
 		if err != nil {
@@ -233,7 +233,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b
 			return nil
 		}
 
-		sendBatch.Commit(out, remote, hostinfo.localIndexId)
+		sendBatch.Commit(out, remote)
 		return nil
 	})
 	if err != nil {
@@ -612,7 +612,8 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
 					"udpAddr", hr,
 				)
 			}
-			if udp.IsUnreachable(err) {
+			// A closing tunnel is about to be torn down, there is no point looking for another path for it
+			if t != header.CloseTunnel && udp.IsUnreachable(err) {
 				hostinfo.remoteUnreachable(f, hr)
 			}
 		}

interface.go

@@ -1,6 +1,7 @@
 package nebula
 
 import (
+	"bytes"
 	"context"
 	"crypto/fips140"
 	"errors"
@@ -507,24 +508,40 @@ func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) {
 		f.metricTxDropped.Inc(int64(dropped))
 	}
 
-	// Packets are tagged with the local index of the tunnel they belong to. A tunnel's packets tend to fail together,
-	// so skip straight past a repeat of the one before.
+	// A tunnel's packets tend to fail together, so skip straight past a repeat of the one before. The first 8 bytes
+	// of a header say whose packet it is, the counter after them changes with every packet.
 	var last batch.Failure
 	for i, fail := range sb.Failures() {
-		if i > 0 && fail.Tag == last.Tag && fail.Dst == last.Dst {
+		if i > 0 && fail.Dst == last.Dst && bytes.Equal(fail.Head[:8], last.Head[:8]) {
 			continue
 		}
 		last = fail
 
-		if !udp.IsUnreachable(fail.Err) {
-			continue
-		}
-		if hostinfo := f.hostMap.QueryIndex(fail.Tag); hostinfo != nil {
-			hostinfo.remoteUnreachable(f, fail.Dst)
+		if udp.IsUnreachable(fail.Err) {
+			f.sendUnreachable(fail.Head[:], fail.Dst)
 		}
 	}
 }
 
+// sendUnreachable finds the tunnel a packet we had no route for was sent on, from the start of the packet, and has it
+// look for another path to its peer.
+func (f *Interface) sendUnreachable(head []byte, dst netip.AddrPort) {
+	var h header.H
+	if h.Parse(head) != nil {
+		return
+	}
+
+	if h.Type == header.Message && h.Subtype == header.MessageRelay {
+		// TODO: a relayed packet's header carries the relay session's index, not the relay tunnel's, so it does not
+		// lead us to the tunnel whose remote failed. The relay tunnel's own traffic finds out soon enough.
+		return
+	}
+
+	if hostinfo := f.hostMap.QueryReverseIndex(h.RemoteIndex); hostinfo != nil {
+		hostinfo.remoteUnreachable(f, dst)
+	}
+}
+
 func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
 	c.RegisterReloadCallback(f.reloadFirewall)
 	for _, reload := range []func(*config.C){

overlay/batch/tx_batch.go

@@ -4,16 +4,20 @@ import "net/netip"
 
 const SendBatchCap = 128
 
+// FailureHeadLen is how much of a refused packet a Failure keeps, nebula's header, which says whose packet it was
+const FailureHeadLen = 16
+
 // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush.
 type batchWriter interface {
 	WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error)
 }
 
-// Failure is a packet the underlay would not send, with the tag it was committed under
+// Failure is a packet the underlay would not send
 type Failure struct {
-	Tag uint32
-	Dst netip.AddrPort
-	Err error
+	// Head is the start of the packet, zero padded if the packet was shorter
+	Head [FailureHeadLen]byte
+	Dst  netip.AddrPort
+	Err  error
 }
 
 // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch.
@@ -23,7 +27,6 @@ type SendBatch struct {
 	out   batchWriter
 	bufs  [][]byte
 	dsts  []netip.AddrPort
-	tags  []uint32
 	arena *Arena
 
 	failures []Failure
@@ -37,7 +40,6 @@ func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch {
 		out:   out,
 		bufs:  make([][]byte, 0, batchCap),
 		dsts:  make([]netip.AddrPort, 0, batchCap),
-		tags:  make([]uint32, 0, batchCap),
 		arena: NewArena(arenaSize),
 	}
 	b.recordFailure = b.failed
@@ -45,7 +47,9 @@ func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch {
 }
 
 func (b *SendBatch) failed(i int, err error) {
-	b.failures = append(b.failures, Failure{Tag: b.tags[i], Dst: b.dsts[i], Err: err})
+	f := Failure{Dst: b.dsts[i], Err: err}
+	copy(f.Head[:], b.bufs[i])
+	b.failures = append(b.failures, f)
 }
 
 func (b *SendBatch) Reserve(sz int) []byte {
@@ -57,12 +61,9 @@ func (b *SendBatch) Reserve(sz int) []byte {
 // bounding how long the first packet of a large read batch waits.
 func (b *SendBatch) Len() int { return len(b.bufs) }
 
-// Commit queues pkt for dst. tag comes back in Failures if the underlay refuses the packet, so the caller can tell
-// whose packet it was.
-func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort, tag uint32) {
+func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort) {
 	b.bufs = append(b.bufs, pkt)
 	b.dsts = append(b.dsts, dst)
-	b.tags = append(b.tags, tag)
 }
 
 // Flush writes every queued packet and reports how many actually went out. A short count means some destinations
@@ -79,7 +80,6 @@ func (b *SendBatch) Flush() (int, error) {
 	clear(b.bufs)
 	b.bufs = b.bufs[:0]
 	b.dsts = b.dsts[:0]
-	b.tags = b.tags[:0]
 	b.arena.Reset()
 	return written, err
 }

overlay/batch/tx_batch_test.go

@@ -36,7 +36,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) {
 			t.Fatalf("slot %d: cap=%d want 32", i, cap(slot))
 		}
 		pkt := append(slot[:0], byte(i), byte(i+1), byte(i+2))
-		b.Commit(pkt, ap, 0)
+		b.Commit(pkt, ap)
 	}
 	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
@@ -77,7 +77,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) {
 	for i := range 3 {
 		s := b.Reserve(8)
 		pkt := append(s[:0], byte(0xA0+i), byte(0xB0+i))
-		b.Commit(pkt, ap, 0)
+		b.Commit(pkt, ap)
 	}
 	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
@@ -98,11 +98,11 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) {
 
 	s1 := b.Reserve(4)
 	pkt1 := append(s1[:0], 0x11, 0x22, 0x33, 0x44)
-	b.Commit(pkt1, ap, 0)
+	b.Commit(pkt1, ap)
 
 	s2 := b.Reserve(8) // exceeds remaining cap, triggers grow
 	pkt2 := append(s2[:0], 0xA, 0xB, 0xC, 0xD, 0xE)
-	b.Commit(pkt2, ap, 0)
+	b.Commit(pkt2, ap)
 
 	// pkt1 must still be intact even though backing reallocated.
 	if pkt1[0] != 0x11 || pkt1[3] != 0x44 {
@@ -140,11 +140,24 @@ func (w *failingBatchWriter) WriteBatch(bufs [][]byte, _ []netip.AddrPort, faile
 	return written, nil
 }
 
-func TestSendBatchReportsFailedTags(t *testing.T) {
+func TestSendBatchReportsFailures(t *testing.T) {
 	errNoRoute := errors.New("no route")
 	fw := &failingBatchWriter{refuse: map[byte]error{2: errNoRoute, 3: errNoRoute}}
-
-	b := NewSendBatch(fw, 4, 64)
+	b := NewSendBatch(fw, 4, 128)
+
+	// pkt is n bytes starting with the refuse key, each later byte its offset plus 0x10 so the head is recognizable
+	pkt := func(key byte, n int) []byte {
+		p := b.Reserve(n)[:n]
+		p[0] = key
+		for i := 1; i < n; i++ {
+			p[i] = byte(0x10 + i)
+		}
+		return p
+	}
+	head := func(p []byte) (h [FailureHeadLen]byte) {
+		copy(h[:], p)
+		return h
+	}
 
 	dsts := []netip.AddrPort{
 		netip.MustParseAddrPort("10.0.0.1:4242"),
@@ -152,10 +165,12 @@ func TestSendBatchReportsFailedTags(t *testing.T) {
 		netip.MustParseAddrPort("[fd00::3]:4242"),
 		netip.MustParseAddrPort("10.0.0.4:4242"),
 	}
-	for i, dst := range dsts {
-		pkt := append(b.Reserve(8)[:0], byte(i+1))
-		b.Commit(pkt, dst, uint32(100+i))
-	}
+	long := pkt(2, FailureHeadLen+8) // only its head is kept
+	short := pkt(3, 4)               // zero padded
+	b.Commit(pkt(1, 20), dsts[0])
+	b.Commit(long, dsts[1])
+	b.Commit(short, dsts[2])
+	b.Commit(pkt(1, 20), dsts[3])
 
 	written, err := b.Flush()
 	if err != nil {
@@ -164,22 +179,23 @@ func TestSendBatchReportsFailedTags(t *testing.T) {
 	if written != 2 {
 		t.Errorf("written = %d, want 2", written)
 	}
-	want := []Failure{{101, dsts[1], errNoRoute}, {102, dsts[2], errNoRoute}}
+	want := []Failure{{head(long), dsts[1], errNoRoute}, {head(short), dsts[2], errNoRoute}}
 	if got := b.Failures(); !slices.Equal(got, want) {
 		t.Errorf("failures = %v, want %v", got, want)
 	}
 
-	// Tags do not leak into the next flush
-	b.Commit(append(b.Reserve(8)[:0], 2), dsts[0], 7)
+	// Failures do not leak into the next flush
+	again := pkt(2, 20)
+	b.Commit(again, dsts[0])
 	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
 	}
-	if want := []Failure{{7, dsts[0], errNoRoute}}; !slices.Equal(b.Failures(), want) {
+	if want := []Failure{{head(again), dsts[0], errNoRoute}}; !slices.Equal(b.Failures(), want) {
 		t.Errorf("second flush failures = %v, want %v", b.Failures(), want)
 	}
 
 	// A clean flush forgets the last one's failures
-	b.Commit(append(b.Reserve(8)[:0], 1), dsts[0], 8)
+	b.Commit(pkt(1, 20), dsts[0])
 	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
 	}

punchy_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/header"
 	"github.com/slackhq/nebula/test"
 	"github.com/slackhq/nebula/udp"
 	"github.com/stretchr/testify/assert"
@@ -248,18 +249,37 @@ func TestPunchyReloadRestartRequired(t *testing.T) {
 // errNoRoute is the unreachable error every platform classifies the same way
 var errNoRoute = udp.ErrInvalidIPv6RemoteForSocket
 
-// recordingConn records where every write went and fails the ones to an address in fail
+// recordingConn records where every write went and what kind of message it was, and fails the ones to an address in
+// fail
 type recordingConn struct {
 	udp.NoopConn
 	fail  map[netip.AddrPort]error
 	wrote []netip.AddrPort
+	// types is the message type of each write, a punch has none
+	types []header.MessageType
 }
 
-func (c *recordingConn) WriteTo(_ []byte, addr netip.AddrPort) error {
+func (c *recordingConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	c.wrote = append(c.wrote, addr)
+	var h header.H
+	if h.Parse(b) == nil {
+		c.types = append(c.types, h.Type)
+	}
 	return c.fail[addr]
 }
 
+func (c *recordingConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(int, error)) (int, error) {
+	written := 0
+	for i := range bufs {
+		if err := c.WriteTo(bufs[i], addrs[i]); err != nil {
+			failed(i, err)
+			continue
+		}
+		written++
+	}
+	return written, nil
+}
+
 type noLighthouses struct{}
 
 func (noLighthouses) IsAnyLighthouseAddr([]netip.Addr) bool { return false }

udp/udp_tester.go

@@ -129,6 +129,14 @@ func (u *TesterConn) SetDualStackAddrs(v4, v6 netip.AddrPort) {
 	u.addrs.Store(a)
 }
 
+// GetAddrFor returns the underlay address a send to dst leaves from, GetAddr when there is no route to dst
+func (u *TesterConn) GetAddrFor(dst netip.AddrPort) netip.AddrPort {
+	if src, err := u.sourceFor(dst); err == nil {
+		return src
+	}
+	return u.GetAddr()
+}
+
 // sourceFor returns the address a send to dst leaves from
 func (u *TesterConn) sourceFor(dst netip.AddrPort) (netip.AddrPort, error) {
 	a := u.addrs.Load()