Compare commits

..

3 Commits

Author SHA1 Message Date
Nate Brown 983784bc41 Group the HostInfo fields the packet paths touch 2026-07-23 16:52:09 -05:00
Nate Brown 30e2735af3 Fold the rebind counter and traffic flags into one atomic word 2026-07-23 16:52:08 -05:00
Nate Brown bce0b4249e Add an e2e Drop exit type and a roaming recovery measurement 2026-07-23 16:52:08 -05:00
7 changed files with 21 additions and 127 deletions
+1 -6
View File
@@ -108,12 +108,7 @@ func (cm *connectionManager) In(h *HostInfo) {
h.markIn() h.markIn()
} }
// OutRelay records relayed traffic, leaving the rebind epoch for the direct path to this host to consume // Out records outbound traffic and reports whether the local network changed since this tunnel last sent.
func (cm *connectionManager) OutRelay(h *HostInfo) {
h.markOutOnly()
}
// Out records outbound traffic and reports whether we rebound since this tunnel last sent
func (cm *connectionManager) Out(h *HostInfo) bool { func (cm *connectionManager) Out(h *HostInfo) bool {
return h.markOut(cm.intf.rebindEpoch.Load()) return h.markOut(cm.intf.rebindEpoch.Load())
} }
-10
View File
@@ -123,16 +123,6 @@ func (c *Control) SetLocalAddrsFn(fn func(*LocalAllowList) []netip.Addr) {
c.f.lightHouse.localAddrsFn = fn c.f.lightHouse.localAddrsFn = fn
} }
// GetRebindEpochFor returns the rebind epoch a tunnel last sent under, so a test can tell whether a send
// consumed the epoch edge without having to infer it from lighthouse traffic.
func (c *Control) GetRebindEpochFor(vpnAddr netip.Addr) (uint32, bool) {
h := c.f.hostMap.QueryVpnAddr(vpnAddr)
if h == nil {
return 0, false
}
return h.state.Load() >> stateEpochShift, true
}
func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool { func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool {
hostinfo := c.f.handshakeManager.QueryVpnAddr(vpnIp) hostinfo := c.f.handshakeManager.QueryVpnAddr(vpnIp)
if hostinfo == nil { if hostinfo == nil {
-54
View File
@@ -223,57 +223,3 @@ func TestRebindAdvertisesNewAddressAfterMove(t *testing.T) {
lhControl.Stop() lhControl.Stop()
myControl.Stop() myControl.Stop()
} }
// A relayed send records traffic but must not consume the rebind epoch. If it does, the next direct send to the
// relay host sees the epoch already current and never requeries, so the far side is never told to punch at our
// new address. This pins the SendVia call site, which the unit tests cannot reach.
func TestRebindRequeriesAfterRelayedSend(t *testing.T) {
t.Parallel()
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
// No lighthouse on purpose: it would hand out a direct address for them and nothing would relay.
myControl, myVpnIpNet, _, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.1/24", m{"relay": m{"use_relays": true}})
relayControl, relayVpnIpNet, relayUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "relay", "10.128.0.128/24", m{"relay": m{"am_relay": true}})
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.2/24", m{"relay": m{"use_relays": true}})
myControl.InjectLightHouseAddr(relayVpnIpNet[0].Addr(), relayUdpAddr)
myControl.InjectRelays(theirVpnIpNet[0].Addr(), []netip.Addr{relayVpnIpNet[0].Addr()})
relayControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
r := router.NewR(t, myControl, relayControl, theirControl)
defer r.RenderFlow()
myControl.Start()
relayControl.Start()
theirControl.Start()
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("establish")))
r.RouteForAllUntilTxTun(theirControl)
r.RouteFor(time.Millisecond * 500)
hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
require.NotNil(t, hi, "expected a tunnel to them")
require.NotEmpty(t, hi.CurrentRelaysToMe, "them must be reachable only via the relay for this test to mean anything")
// sendNoMetrics only reaches SendVia when there is no direct remote, so pin that too. Without this the test
// keeps passing while quietly sending direct and never exercising the relay path.
require.False(t, hi.CurrentRemote.IsValid(), "them must have no direct remote, otherwise SendVia is never called")
before, ok := myControl.GetRebindEpochFor(relayVpnIpNet[0].Addr())
require.True(t, ok, "expected a tunnel to the relay")
myControl.RebindUDPServer()
// Traffic to them goes through SendVia on the relay tunnel. That must record traffic without consuming the
// relay tunnel's own epoch edge, which belongs to the direct path.
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("relayed")))
r.RouteForAllUntilTxTun(theirControl)
after, ok := myControl.GetRebindEpochFor(relayVpnIpNet[0].Addr())
require.True(t, ok)
assert.Equal(t, before, after,
"a relayed send consumed the relay tunnel's rebind epoch, so the next direct send will not requery")
myControl.Stop()
relayControl.Stop()
theirControl.Stop()
}
+14 -14
View File
@@ -238,12 +238,16 @@ const (
) )
type HostInfo struct { type HostInfo struct {
// The first cache line is everything the packet paths touch. // The first cache line is everything the packet paths touch. Grouping them here means a send or receive
// pulls in one line instead of two, which is what the layout looked like when state lived at the end.
remote atomic.Pointer[netip.AddrPort] remote atomic.Pointer[netip.AddrPort]
ConnectionState *ConnectionState ConnectionState *ConnectionState
// Traffic bits, pendingDeletion, and the rebind epoch we last sent under // state holds everything the hot paths need to touch per packet, in one word: whether we have seen traffic
// each way since the connection manager last looked, whether it has given up on us, and the
// Interface.rebindEpoch this tunnel last sent under. Keeping the epoch here means it survives the traffic
// bits being cleared, so a tunnel that has not sent since a rebind still notices when it does.
state atomic.Uint32 state atomic.Uint32
promoteCounter atomic.Uint32 promoteCounter atomic.Uint32
@@ -759,7 +763,7 @@ func (i *HostInfo) TryPromoteBest(preferredRanges []netip.Prefix, ifce *Interfac
} }
} }
// Bits within HostInfo.state, everything above stateEpochShift is the epoch // Bits within HostInfo.state. Everything above stateEpochShift is the rebind epoch.
const ( const (
stateIn uint32 = 1 << iota stateIn uint32 = 1 << iota
stateOut stateOut
@@ -769,14 +773,17 @@ const (
stateEpochShift = 3 stateEpochShift = 3
) )
// markIn records inbound traffic // markIn records inbound traffic. Reading first keeps the cache line shared on the common path, where the bit
// is already set.
func (i *HostInfo) markIn() { func (i *HostInfo) markIn() {
if i.state.Load()&stateIn == 0 { if i.state.Load()&stateIn == 0 {
i.state.Or(stateIn) i.state.Or(stateIn)
} }
} }
// markOut records a send and reports whether the epoch moved, meaning we want a punch from the far side // markOut records that we sent on this tunnel under the given rebind epoch. It reports whether the epoch moved
// since our last send, which means the local network changed and we want the far side to punch at us again.
// The common path is a single load that matches and returns.
func (i *HostInfo) markOut(epoch uint32) bool { func (i *HostInfo) markOut(epoch uint32) bool {
e := epoch << stateEpochShift e := epoch << stateEpochShift
for { for {
@@ -791,19 +798,12 @@ func (i *HostInfo) markOut(epoch uint32) bool {
} }
} }
// markOutOnly records a send without consuming the rebind epoch, for paths that cannot act on a requery // sentSinceCheck reports whether anything has been sent since the connection manager last looked.
func (i *HostInfo) markOutOnly() {
if i.state.Load()&stateOut == 0 {
i.state.Or(stateOut)
}
}
// sentSinceCheck reports whether anything has been sent since the connection manager last looked
func (i *HostInfo) sentSinceCheck() bool { func (i *HostInfo) sentSinceCheck() bool {
return i.state.Load()&stateOut != 0 return i.state.Load()&stateOut != 0
} }
// takeTraffic clears both traffic bits, leaving the epoch alone, and reports what they were // takeTraffic clears both traffic bits, leaving the epoch alone, and reports what they were.
func (i *HostInfo) takeTraffic() (in bool, out bool) { func (i *HostInfo) takeTraffic() (in bool, out bool) {
old := i.state.And(^(stateIn | stateOut)) old := i.state.And(^(stateIn | stateOut))
return old&stateIn != 0, old&stateOut != 0 return old&stateIn != 0, old&stateOut != 0
-40
View File
@@ -401,43 +401,3 @@ func TestHostMap_RelayState(t *testing.T) {
assert.Equal(t, []netip.Addr{}, h1.relayState.relays) assert.Equal(t, []netip.Addr{}, h1.relayState.relays)
} }
func TestHostInfo_markOut(t *testing.T) {
h := &HostInfo{}
h.markOut(5) // stamped when the tunnel was added
// A tunnel already on the current epoch has nothing to report, which is what keeps a fresh tunnel from
// requerying on its first packet
assert.False(t, h.markOut(5), "an unchanged epoch should not report a move")
assert.True(t, h.sentSinceCheck(), "the send is still recorded as traffic")
// A rebind is observed exactly once, so we requery once per rebind
assert.True(t, h.markOut(6), "a bumped epoch should report a move")
assert.False(t, h.markOut(6), "the epoch move should only be reported once")
// Traffic and pendingDeletion live in the same word and must survive an epoch change
h.setPendingDeletion(true)
h.markIn()
assert.True(t, h.markOut(7))
assert.True(t, h.isPendingDeletion(), "pendingDeletion must survive an epoch change")
in, out := h.takeTraffic()
assert.True(t, in, "inbound traffic must survive an epoch change")
assert.True(t, out)
// Clearing the traffic bits leaves the epoch alone, otherwise an idle tunnel would requery forever
assert.False(t, h.markOut(7), "takeTraffic must not disturb the epoch")
}
// A relayed send records traffic but must leave the rebind epoch for the direct path to consume, otherwise
// relaying to a host swallows the requery that gets the far side punching at our new address.
func TestHostInfo_markOutOnly(t *testing.T) {
h := &HostInfo{}
h.markOut(5)
h.markOutOnly()
assert.True(t, h.sentSinceCheck(), "a relayed send is still outbound traffic")
assert.False(t, h.markOut(5), "a relayed send must not disturb the epoch")
assert.True(t, h.markOut(6), "a relayed send must not consume the epoch edge")
assert.False(t, h.markOut(6))
}
+3 -2
View File
@@ -297,7 +297,7 @@ func (f *Interface) SendVia(via *HostInfo,
c := via.ConnectionState.messageCounter.Add(1) c := via.ConnectionState.messageCounter.Add(1)
out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c) out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c)
f.connectionManager.OutRelay(via) f.connectionManager.Out(via)
// Authenticate the header and payload, but do not encrypt for this message type. // Authenticate the header and payload, but do not encrypt for this message type.
// The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload. // The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload.
@@ -365,7 +365,8 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
//l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p) //l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p)
out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c) out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
// We rebound since this tunnel last sent, ask the lighthouse to get the far side punching at us again // We rebound since this tunnel last sent, so the local network moved. Ask the lighthouse to have the far side
// punch at where we are now, which primes their conntrack the same way a handshake would.
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel { if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0]) f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
if f.l.Enabled(context.Background(), slog.LevelDebug) { if f.l.Enabled(context.Background(), slog.LevelDebug) {
+3 -1
View File
@@ -82,7 +82,9 @@ type Interface struct {
sendRecvErrorConfig recvErrorConfig sendRecvErrorConfig recvErrorConfig
acceptRecvErrorConfig recvErrorConfig acceptRecvErrorConfig recvErrorConfig
// Bumped on every udp rebind, tunnels compare it to decide they need a punch from the far side // rebindEpoch bumps every time the udp listener is rebound, which means the local network moved. Tunnels
// compare it against their own copy to decide they need a punch from the far side. Read on every send, only
// written on a rebind, so the cache line stays shared across the routines.
rebindEpoch atomic.Uint32 rebindEpoch atomic.Uint32
version string version string