mapper: deliver peer removals as separate deltas

Updates #3492

(cherry picked from commit eeaac680be)
This commit is contained in:
Kristoffer Dalby committed 2026-10-13 14:35:52 +02:00
1 parent 929e3337f4
commit 98ce49defd
20 files changed
+602 -146

No files matched your search

+4 -5
View File
@@ -14,6 +14,7 @@ import (
"os/signal"
"path/filepath"
"runtime"
"slices"
"strings"
"sync"
"syscall"
@@ -176,10 +177,8 @@ func NewHeadscale(cfg *types.Config) (*Headscale, error) {
return
}
policyChanged, err := app.state.DeleteNode(node)
if !policyChanged.IsEmpty() {
app.Change(policyChanged)
}
changes, err := app.state.DeleteNode(node)
app.Change(changes...)
if err != nil {
log.Error().Err(err).EmbedObject(node).Msg("ephemeral node deletion failed")
@@ -1168,7 +1167,7 @@ func readOrCreatePrivateKey(path string) (*key.MachinePrivate, error) {
// All change should be enqueued here and empty will be automatically
// ignored.
func (h *Headscale) Change(cs ...change.Change) {
h.mapBatcher.AddWork(cs...)
h.mapBatcher.AddWork(slices.Concat(cs, h.state.DrainSelfRefreshes())...)
}
// HTTPHandler returns an [http.Handler] for the [Headscale] control server.
+2 -4
View File
@@ -217,10 +217,8 @@ func (h *Headscale) handleLogout(
EmbedObject(node).
Msg("Deleting ephemeral node during logout")
c, err := h.state.DeleteNode(node)
if !c.IsEmpty() {
h.Change(c)
}
changes, err := h.state.DeleteNode(node)
h.Change(changes...)
if err != nil {
return nil, fmt.Errorf("deleting ephemeral node: %w", err)
+9 -1
View File
@@ -15,14 +15,22 @@ import (
// minVersionParts is the minimum number of version parts needed for major.minor.
const minVersionParts = 2
// fullNetmapRemovalsCapVer is the capability version of Tailscale main when
// clients started reporting peers a full netmap drops as removed
// (tailscale/tailscale#15660); v1.104 is the first release at or above it.
const fullNetmapRemovalsCapVer tailcfg.CapabilityVersion = 148
// CanOldCodeBeCleanedUp is called at server startup to panic when
// [MinSupportedCapabilityVersion] has crossed a threshold at which a
// backwards-compat emit path can be deleted. Each entry pairs a
// [tailcfg.CapabilityVersion] threshold with the message identifying
// the code to remove; today there are none.
// the code to remove.
//
// All capability-version-gated cleanups should be registered here.
func CanOldCodeBeCleanedUp() {
if MinSupportedCapabilityVersion >= fullNetmapRemovalsCapVer {
panic("the tailscale/tailscale#15660 compat can be cleaned up: grep for it (hscontrol/mapper/batcher.go and its tests)")
}
}
func tailscaleVersSorted() []string {
+2 -4
View File
@@ -438,10 +438,8 @@ func (api headscaleV1APIServer) DeleteNode(
return nil, status.Errorf(codes.NotFound, "node not found")
}
nodeChange, err := api.h.state.DeleteNode(node)
if !nodeChange.IsEmpty() {
api.h.Change(nodeChange)
}
changes, err := api.h.state.DeleteNode(node)
api.h.Change(changes...)
if err != nil {
return nil, err
+97 -60
View File
@@ -78,13 +78,18 @@ type nodeConnection interface {
updateSentPeers(resp *tailcfg.MapResponse)
}
// generateMapResponse generates a [tailcfg.MapResponse] for the given [types.NodeID] based on the provided [change.Change].
func generateMapResponse(nc nodeConnection, mapper *mapper, r change.Change) (*tailcfg.MapResponse, error) {
// generateMapResponse generates the [tailcfg.MapResponse] values to send, in
// order, for the given [types.NodeID] based on the provided [change.Change].
func generateMapResponse(
nc nodeConnection,
mapper *mapper,
r change.Change,
) ([]*tailcfg.MapResponse, error) {
nodeID := nc.nodeID()
version := nc.version()
if r.IsEmpty() {
return nil, nil //nolint:nilnil // Empty response means nothing to send
return nil, nil
}
if nodeID == 0 {
@@ -97,7 +102,7 @@ func generateMapResponse(nc nodeConnection, mapper *mapper, r change.Change) (*t
// Handle self-only responses
if r.IsSelfOnly() && r.TargetNode != nodeID {
return nil, nil //nolint:nilnil // No response needed for other nodes when self-only
return nil, nil
}
// Check if this is a self-update (the changed node is the receiving node).
@@ -106,7 +111,8 @@ func generateMapResponse(nc nodeConnection, mapper *mapper, r change.Change) (*t
isSelfUpdate := r.OriginNode != 0 && r.OriginNode == nodeID
var (
mapResp *tailcfg.MapResponse
resp *tailcfg.MapResponse
removed []tailcfg.NodeID
err error
)
@@ -122,51 +128,60 @@ func generateMapResponse(nc nodeConnection, mapper *mapper, r change.Change) (*t
currentPeerIDs = append(currentPeerIDs, peer.ID().NodeID())
}
removedPeers := nc.computePeerDiff(currentPeerIDs)
// Routes and policy changes also affect self. Each connection drops
// the self node when it already holds the same value.
mapResp, err = mapper.policyChangeResponse(nodeID, version, removedPeers, currentPeers, true)
removed = nc.computePeerDiff(currentPeerIDs)
// Include self node when this is a self-update (e.g., node's own tags changed)
// so the node sees its updated self info along with new packet filters.
resp, err = mapper.policyChangeResponse(nodeID, version, currentPeers, true)
} else if isSelfUpdate {
// Non-policy self-update: just send the self node info
mapResp, err = mapper.selfMapResponse(nodeID, version)
resp, err = mapper.selfMapResponse(nodeID, version)
} else {
mapResp, err = mapper.buildFromChange(nodeID, version, &r)
resp, err = mapper.buildFromChange(nodeID, version, &r)
}
if err != nil {
return nil, fmt.Errorf("generating map response for nodeID %d: %w", nodeID, err)
}
// When a full update (SendAllPeers=true) produces zero visible peers
// (e.g., a restrictive policy isolates this node), the resulting
// [tailcfg.MapResponse] has Peers: []*tailcfg.Node{} (empty non-nil slice).
//
// The Tailscale client only treats Peers as a full authoritative
// replacement when len(Peers) > 0 (controlclient/map.go:462).
// An empty Peers slice is indistinguishable from a delta response,
// so the client silently preserves its existing peer state.
//
// This matters when a [change.FullUpdate] replaces a pending
// [change.PolicyChange] in the batcher ([Batcher.addToBatch]
// short-circuits on [change.HasFull]). The [change.PolicyChange]
// would have computed PeersRemoved via
// [multiChannelNodeConn.computePeerDiff], but the [change.FullUpdate]
// path uses [MapResponseBuilder.WithPeers] which sets Peers: [].
//
// Fix: when a full update results in zero peers, compute the diff
// against lastSentPeers and add explicit PeersRemoved entries so
// the client correctly clears its stale peer state.
if mapResp != nil && r.SendAllPeers && len(mapResp.Peers) == 0 {
removedPeers := nc.computePeerDiff(nil)
if len(removedPeers) > 0 {
mapResp.PeersRemoved = removedPeers
}
if resp == nil {
return nil, nil
}
return mapResp, nil
// A full peer list replaces the node's peer set, so peers missing from
// it are removals. Clients take Peers as authoritative only when it is
// non-empty, so a full update that isolates a node (e.g.
// [Batcher.addToBatch] replacing a pending [change.PolicyChange] with a
// [change.FullUpdate]) needs them sent explicitly.
if resp.Peers != nil {
peerIDs := make([]tailcfg.NodeID, 0, len(resp.Peers))
for _, peer := range resp.Peers {
peerIDs = append(peerIDs, peer.ID)
}
removed = nc.computePeerDiff(peerIDs)
}
return withRemovals(resp, removed), nil
}
// handleNodeChange generates and sends a [tailcfg.MapResponse] for a given node and [change.Change].
// withRemovals returns resp and the peers removed from the node's view as the
// responses to send. The removals lead, in a response of their own: clients
// before Tailscale v1.104 only report removed peers to IPN bus watchers opted
// out of full netmaps (the Android app since 1.100) from responses they apply
// as deltas, and resp can carry DNSConfig, SSHPolicy, Node or Peers, which
// force a full rebuild.
//
// TODO(kradalby): when the tailscale/tailscale#15660 compat goes (see
// capver.CanOldCodeBeCleanedUp), set resp.PeersRemoved and return resp alone.
func withRemovals(resp *tailcfg.MapResponse, removed []tailcfg.NodeID) []*tailcfg.MapResponse {
if len(removed) == 0 {
return []*tailcfg.MapResponse{resp}
}
return []*tailcfg.MapResponse{{PeersRemoved: removed}, resp}
}
// handleNodeChange generates and sends the [tailcfg.MapResponse] values for a given node and [change.Change].
func handleNodeChange(nc nodeConnection, mapper *mapper, r change.Change) error {
if nc == nil {
return ErrNodeConnectionNil
@@ -176,35 +191,31 @@ func handleNodeChange(nc nodeConnection, mapper *mapper, r change.Change) error
log.Debug().Caller().Uint64(zf.NodeID, nodeID.Uint64()).Str(zf.Reason, r.Reason).Msg("node change processing started")
data, err := generateMapResponse(nc, mapper, r)
resps, err := generateMapResponse(nc, mapper, r)
if err != nil {
return fmt.Errorf("generating map response for node %d: %w", nodeID, err)
}
if data == nil {
// No data to send is valid for some response types
return nil
}
for _, resp := range resps {
err = nc.send(resp)
if err != nil {
// If the node has no active connections, the data was not
// delivered. Do not update lastSentPeers — recording phantom
// peer state would corrupt future computePeerDiff calculations,
// causing the node to miss peer additions or removals after
// reconnection.
if errors.Is(err, errNoActiveConnections) {
return nil
}
// Send the map response
err = nc.send(data)
if err != nil {
// If the node has no active connections, the data was not
// delivered. Do not update lastSentPeers — recording phantom
// peer state would corrupt future computePeerDiff calculations,
// causing the node to miss peer additions or removals after
// reconnection.
if errors.Is(err, errNoActiveConnections) {
return nil
return fmt.Errorf("sending map response to node %d: %w", nodeID, err)
}
return fmt.Errorf("sending map response to node %d: %w", nodeID, err)
// Update peer tracking only after confirmed delivery to at
// least one active connection.
nc.updateSentPeers(resp)
}
// Update peer tracking only after confirmed delivery to at
// least one active connection.
nc.updateSentPeers(data)
return nil
}
@@ -341,6 +352,29 @@ func (b *Batcher) AddNode(
// path, and under workMu so a concurrent async bundle for this node
// cannot interleave its own lastSentPeers update.
nodeConn.workMu.Lock()
// Peers removed while the node had no stream: the initial map is a
// full rebuild, so follow it with the removal as a delta, see
// [withRemovals].
//
// TODO(kradalby): delete with the tailscale/tailscale#15660 compat, see
// capver.CanOldCodeBeCleanedUp.
peerIDs := make([]tailcfg.NodeID, 0, len(initialMap.Peers))
for _, peer := range initialMap.Peers {
peerIDs = append(peerIDs, peer.ID)
}
removed := make([]types.NodeID, 0)
for _, peerID := range nodeConn.computePeerDiff(peerIDs) {
removed = append(removed, types.NodeID(peerID)) //nolint:gosec // NodeID types are equivalent
}
if len(removed) > 0 {
rm := change.PeersRemoved(removed...)
rm.TargetNode = id
b.AddWork(rm)
}
nodeConn.updateSentPeers(initialMap)
nodeConn.workMu.Unlock()
@@ -525,9 +559,12 @@ func (b *Batcher) worker(workerID int) {
// node waits until the initial map is sent.
nc.workMu.Lock()
var err error
result.mapResponse, err = generateMapResponse(nc, b.mapper, w.changes[0])
resps, err := generateMapResponse(nc, b.mapper, w.changes[0])
if len(resps) > 0 {
// The initial map must be the stream's first frame; AddNode
// sends any removals ahead of it after it instead.
result.mapResponse = resps[len(resps)-1]
}
result.err = err
if result.err != nil {
+116 -3
View File
@@ -19,6 +19,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"tailscale.com/tailcfg"
"tailscale.com/types/netmap"
)
var errNodeNotFoundAfterAdd = errors.New("node not found after adding to batcher")
@@ -2159,9 +2160,8 @@ func TestNodeDeletedWhileChangesPending(t *testing.T) {
// Delete the node from state - this returns a NodeRemoved change
// In production, this change is sent to batcher via app.Change()
nodeChange, err := st.DeleteNode(nodeToDelete)
nodeChanges, err := st.DeleteNode(nodeToDelete)
require.NoError(t, err, "should be able to delete node from state")
t.Logf("Deleted node %d from state, change: %s", node3.n.ID, nodeChange.Reason)
// Verify node is deleted from state
_, exists := st.GetNodeByID(node3.n.ID)
@@ -2169,7 +2169,7 @@ func TestNodeDeletedWhileChangesPending(t *testing.T) {
// Send the NodeRemoved change to batcher (this is what app.Change() does)
// With the fix, this should clean up node3 from batcher's internal state
batcher.AddWork(nodeChange)
batcher.AddWork(nodeChanges...)
// Wait for the batcher to process the removal and clean up the node
assert.EventuallyWithT(t, func(c *assert.CollectT) {
@@ -2415,3 +2415,116 @@ func TestSelfSentOnlyWhenChanged(t *testing.T) {
assert.Contains(t, frames[0].Node.Name, "renamed-again")
assert.Nil(t, frames[1].Node, "second connection already holds the new self")
}
// TestHandleNodeChangeSendsRemovalsAsDelta checks that peers missing from a
// response listing the node's complete peer set go out first as a response
// of their own that clients can apply as a delta, and that the full-set
// response never carries them. Clients only report removals to IPN bus
// watchers opted out of full netmaps from delta responses.
//
// TODO(kradalby): with the tailscale/tailscale#15660 compat gone, removals
// ride the full-set response; adapt this and TestHandleNodeChangeRetryAfterRemoval.
func TestHandleNodeChangeSendsRemovalsAsDelta(t *testing.T) {
tests := []struct {
name string
nodes int
ch change.Change
}{
{"policy_change", 2, change.PolicyChange()},
{"full_update", 2, change.FullUpdate()},
// Clients ignore an empty Peers list, so an isolating full update
// relies on the removal entirely.
{"full_update_isolated", 1, change.FullUpdate()},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, tt.nodes, normalBufferSize)
defer cleanup()
gone := tailcfg.NodeID(999)
mc := newMockNodeConnection(testData.Nodes[0].n.ID)
mc.peers.Store(gone, struct{}{})
for i := range testData.Nodes[1:] {
mc.peers.Store(testData.Nodes[i+1].n.ID.NodeID(), struct{}{})
}
require.NoError(t, handleNodeChange(mc, testData.Batcher.mapper, tt.ch))
sent := mc.getSent()
require.Len(t, sent, 2)
assert.Equal(t, []tailcfg.NodeID{gone}, sent[0].PeersRemoved)
_, ok := netmap.MutationsFromMapResponse(sent[0], time.Time{})
assert.True(t, ok, "removal must be applicable as a delta: %+v", sent[0])
assert.Empty(t, sent[1].PeersRemoved, "full-set response must not carry removals")
_, tracked := mc.peers.Load(gone)
assert.False(t, tracked, "removed peer must leave lastSentPeers")
})
}
}
// TestHandleNodeChangeRetryAfterRemoval checks that a change retried after
// its removal was delivered but its content was not does not repeat the
// removal.
func TestHandleNodeChangeRetryAfterRemoval(t *testing.T) {
testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, 2, normalBufferSize)
defer cleanup()
gone := tailcfg.NodeID(999)
mc := newMockNodeConnection(testData.Nodes[0].n.ID)
mc.peers.Store(gone, struct{}{})
var sends int
mc.sendFn = func(resp *tailcfg.MapResponse) error {
sends++
if sends == 2 {
return errNoReadyConnections
}
mc.sent = append(mc.sent, resp)
return nil
}
err := handleNodeChange(mc, testData.Batcher.mapper, change.PolicyChange())
require.ErrorIs(t, err, errNoReadyConnections)
require.NoError(t, handleNodeChange(mc, testData.Batcher.mapper, change.PolicyChange()))
sent := mc.getSent()
require.Len(t, sent, 2, "removal once, then the retried content")
assert.Equal(t, []tailcfg.NodeID{gone}, sent[0].PeersRemoved)
assert.Empty(t, sent[1].PeersRemoved)
assert.NotEmpty(t, sent[1].PacketFilters)
}
// TestDNSConfigOnlyWithSelfRefresh checks policy responses leave DNSConfig
// out, which clients read as unchanged, while self refreshes carry it: a
// node's DNS config derives from its own CapMap and Hostinfo, and a
// DNSConfig forces clients into a full netmap rebuild.
func TestDNSConfigOnlyWithSelfRefresh(t *testing.T) {
testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, 2, normalBufferSize)
defer cleanup()
testData.Config.TailcfgDNSConfig = &tailcfg.DNSConfig{
Proxied: true,
Domains: []string{"headscale.test"},
}
self := testData.Nodes[0].n.ID
mc := newMockNodeConnection(self)
require.NoError(t, handleNodeChange(mc, testData.Batcher.mapper, change.PolicyChange()))
require.NoError(t, handleNodeChange(mc, testData.Batcher.mapper, change.SelfUpdate(self)))
sent := mc.getSent()
require.Len(t, sent, 2)
assert.Nil(t, sent[0].DNSConfig, "policy response must not carry DNSConfig")
assert.NotNil(t, sent[1].DNSConfig, "self refresh must carry DNSConfig")
}
+6 -20
View File
@@ -318,19 +318,17 @@ func (m *mapper) selfMapResponse(
}
// policyChangeResponse creates a [tailcfg.MapResponse] for policy changes.
// It sends:
// - PeersRemoved for peers that are no longer visible after the policy change
// Peers no longer visible after the change are sent separately, see
// [handleNodeChange]. It sends:
// - PeersChanged for remaining peers (their AllowedIPs may have changed due to policy)
// - Updated PacketFilters
// - Updated SSHPolicy (SSH rules may reference users/groups that changed)
// - DNSConfig so the client's resolver state stays anchored even when a
// policy-triggered wgengine reconfigure races a netmon LinkChange (the
// LinkChange handler reapplies dns.Manager.Set with the engine's
// lastDNSConfig; if that snapshot is stale, the OS resolver loses the
// MagicDNS reverse-DNS routes and Nameservers and curl-by-FQDN stops
// resolving for the rest of the policy window).
// - Optionally, the node's own self info (when includeSelf is true)
//
// DNSConfig is left out: it forces clients into a full netmap rebuild, and
// the node's DNS config inputs arrive with its own [change.SelfUpdate], see
// [state.State.DrainSelfRefreshes].
//
// This avoids the issue where an empty Peers slice is interpreted by Tailscale
// clients as "no change" rather than "no peers".
// When includeSelf is true, the node's self info is included so that a node
@@ -339,14 +337,12 @@ func (m *mapper) selfMapResponse(
func (m *mapper) policyChangeResponse(
nodeID types.NodeID,
capVer tailcfg.CapabilityVersion,
removedPeers []tailcfg.NodeID,
currentPeers views.Slice[types.NodeView],
includeSelf bool,
) (*tailcfg.MapResponse, error) {
builder := m.NewMapResponseBuilder(nodeID).
WithDebugType(policyResponseDebug).
WithCapabilityVersion(capVer).
WithDNSConfig().
WithPacketFilters().
WithSSHPolicy()
@@ -354,16 +350,6 @@ func (m *mapper) policyChangeResponse(
builder = builder.WithSelfNode()
}
if len(removedPeers) > 0 {
// Convert [tailcfg.NodeID] to [types.NodeID] for [MapResponseBuilder.WithPeersRemoved]
removedIDs := make([]types.NodeID, len(removedPeers))
for i, id := range removedPeers {
removedIDs[i] = types.NodeID(id) //nolint:gosec // NodeID types are equivalent
}
builder.WithPeersRemoved(removedIDs...)
}
// Send remaining peers in PeersChanged - their AllowedIPs may have
// changed due to the policy update (e.g., different routes allowed).
// Cross-user peers must also carry their user profile, otherwise the
+13 -4
View File
@@ -10,6 +10,7 @@ import (
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/juanfont/headscale/hscontrol/policy/matcher"
@@ -88,6 +89,9 @@ type PolicyManager struct {
nodeAttrsMap map[types.NodeID]tailcfg.NodeCapMap
nodeAttrsHashes map[types.NodeID]deephash.Sum
nodeAttrsChanged []types.NodeID
// Skip the policy lock when no node-attribute refresh is pending.
nodeAttrsPending atomic.Bool
}
// filterAndPolicy combines the compiled filter rules with policy content for hashing.
@@ -1877,6 +1881,10 @@ func (pm *PolicyManager) refreshNodeAttrsLocked() error {
pm.nodeAttrsHashes = newHashes
pm.nodeAttrsChanged = append(pm.nodeAttrsChanged, changed...)
if len(pm.nodeAttrsChanged) > 0 {
pm.nodeAttrsPending.Store(true)
}
return nil
}
@@ -1924,9 +1932,9 @@ func (pm *PolicyManager) NodeCapMaps() map[types.NodeID]tailcfg.NodeCapMap {
// NodesWithChangedCapMap returns the IDs of nodes whose nodeAttrs
// CapMap shifted across one or more [PolicyManager.updateLocked] calls
// since the last drain. The buffer drains on return. The mapper calls
// this once per [state.State.ReloadPolicy] to decide which nodes need
// a [change.SelfUpdate].
// since the last drain. The buffer drains on return.
// [state.State.DrainSelfRefreshes] calls this whenever changes are
// dispatched to decide which nodes need a [change.SelfUpdate].
//
// [PolicyManager.refreshNodeAttrsLocked] APPENDS to the buffer; the drain
// returns the union of every change since the previous read. A concurrent
@@ -1934,7 +1942,7 @@ func (pm *PolicyManager) NodeCapMaps() map[types.NodeID]tailcfg.NodeCapMap {
// [PolicyManager.SetPolicy] and a drain cannot silently lose the
// policy-reload diff.
func (pm *PolicyManager) NodesWithChangedCapMap() []types.NodeID {
if pm == nil {
if pm == nil || !pm.nodeAttrsPending.Load() {
return nil
}
@@ -1943,6 +1951,7 @@ func (pm *PolicyManager) NodesWithChangedCapMap() []types.NodeID {
out := pm.nodeAttrsChanged
pm.nodeAttrsChanged = nil
pm.nodeAttrsPending.Store(false)
return out
}
+4 -4
View File
@@ -596,9 +596,9 @@ func TestDeletedNodeEndsLongPoll(t *testing.T) {
t.Fatal("expected the initial map write to start")
}
c, err := app.state.DeleteNode(nodeView)
cs, err := app.state.DeleteNode(nodeView)
require.NoError(t, err)
app.Change(c)
app.Change(cs...)
// The reconnect grace period is 10s, so a generous bound here still fails
// if teardown waits for a node that can never come back.
@@ -677,9 +677,9 @@ func TestDeletedNodeInterruptsBlockedWrite(t *testing.T) {
t.Fatal("expected the initial map write to block")
}
c, err := app.state.DeleteNode(nodeView)
cs, err := app.state.DeleteNode(nodeView)
require.NoError(t, err)
app.Change(c)
app.Change(cs...)
select {
case <-serveDone:
+66 -7
View File
@@ -3,6 +3,7 @@ package servertest
import (
"context"
"fmt"
"slices"
"sync"
"testing"
"time"
@@ -16,7 +17,9 @@ import (
"tailscale.com/types/key"
"tailscale.com/types/netmap"
"tailscale.com/types/persist"
"tailscale.com/types/views"
"tailscale.com/util/eventbus"
"tailscale.com/wgengine/filter"
)
// TestClient wraps a Tailscale [controlclient.Direct] connected to a
@@ -42,6 +45,10 @@ type TestClient struct {
netmap *netmap.NetworkMap
history []*netmap.NetworkMap
// deltaUpdates and deltaRemoved: see [WithDeltaUpdates].
deltaUpdates bool
deltaRemoved []tailcfg.NodeID
// updates is a buffered channel that receives a signal
// each time a new [netmap.NetworkMap] arrives.
updates chan *netmap.NetworkMap
@@ -55,10 +62,19 @@ type TestClient struct {
type ClientOption func(*clientConfig)
type clientConfig struct {
ephemeral bool
hostname string
tags []string
user *types.User
ephemeral bool
hostname string
tags []string
user *types.User
deltaUpdates bool
}
// WithDeltaUpdates makes the client accept incremental map updates the way
// a real tailscaled does, and record the peer removals it could apply that
// way ([TestClient.DeltaRemovedPeers]). Only those removals reach IPN bus
// watchers opted out of full netmaps, such as the Android app.
func WithDeltaUpdates() ClientOption {
return func(c *clientConfig) { c.deltaUpdates = true }
}
// WithEphemeral makes the client register as an ephemeral node.
@@ -156,6 +172,8 @@ func NewClient(tb testing.TB, server *TestServer, name string, opts ...ClientOpt
bus: bus,
dialer: dialer,
tracker: tracker,
deltaUpdates: cc.deltaUpdates,
}
tb.Cleanup(func() {
@@ -197,9 +215,13 @@ func (c *TestClient) startPoll(tb testing.TB) {
go func() {
defer close(c.pollDone)
// [controlclient.Direct.PollNetMap] blocks until ctx is cancelled or the server closes
// the connection.
_ = c.direct.PollNetMap(c.pollCtx, c)
var updater controlclient.NetmapUpdater = c
if c.deltaUpdates {
updater = deltaRecorder{c}
}
_ = c.direct.PollNetMap(c.pollCtx, updater)
}()
}
@@ -218,6 +240,43 @@ func (c *TestClient) UpdateFullNetmap(nm *netmap.NetworkMap) {
}
}
// deltaRecorder is the [controlclient.NetmapUpdater] of a [WithDeltaUpdates]
// client. Like tailscaled it takes packet filter and user profile updates
// narrowly, which lets [controlclient.Direct] try the delta path. It
// records removals and declines the mutations, so the full netmap is
// still delivered and the client's peer state stays exact.
type deltaRecorder struct{ *TestClient }
func (d deltaRecorder) UpdatePacketFilter(views.Slice[tailcfg.FilterRule], []filter.Match) bool {
return true
}
func (d deltaRecorder) UpdateUserProfiles(map[tailcfg.UserID]tailcfg.UserProfileView) bool {
return true
}
func (d deltaRecorder) UpdateNetmapDelta(muts []netmap.NodeMutation) bool {
d.mu.Lock()
defer d.mu.Unlock()
for _, m := range muts {
if _, ok := m.(netmap.NodeMutationRemove); ok {
d.deltaRemoved = append(d.deltaRemoved, m.NodeIDBeingMutated())
}
}
return false
}
// DeltaRemovedPeers returns the peers whose removal arrived in a response
// the client could apply incrementally; see [WithDeltaUpdates].
func (c *TestClient) DeltaRemovedPeers() []tailcfg.NodeID {
c.mu.RLock()
defer c.mu.RUnlock()
return slices.Clone(c.deltaRemoved)
}
// cleanup releases all resources.
func (c *TestClient) cleanup() {
if c.pollCancel != nil {
+95 -1
View File
@@ -648,7 +648,7 @@ func TestIssuesServerMutations(t *testing.T) {
deleteChange, err := srv.State().DeleteNode(node2View)
require.NoError(t, err)
srv.App.Change(deleteChange)
srv.App.Change(deleteChange...)
c1.WaitForCondition(t, "deleted peer gone", 10*time.Second,
func(nm *netmap.NetworkMap) bool {
@@ -1087,6 +1087,100 @@ func TestIssuesIdentity(t *testing.T) {
})
}
// TestPeerRemovedAsDelta checks a peer leaving a client's view reaches it
// in a response it can apply as a delta. Bundled with DNSConfig or
// SSHPolicy, the removal forces a full netmap rebuild, which never tells IPN
// bus watchers opted out of full netmaps (the Android app since 1.100) that
// the peer is gone, so they keep showing it.
//
// TODO(kradalby): with the tailscale/tailscale#15660 compat gone, removals
// ride full rebuilds; assert they reach the netmap instead.
func TestPeerRemovedAsDelta(t *testing.T) {
t.Parallel()
// MagicDNS puts DNSConfig in policy responses, making them full rebuilds.
setup := func(t *testing.T) (*servertest.TestServer, *servertest.TestClient, types.NodeID) {
t.Helper()
srv := servertest.NewServer(t, servertest.WithMagicDNS("delta.example.com"))
user1 := srv.CreateUser(t, "delta-user1")
user2 := srv.CreateUser(t, "delta-user2")
c1 := servertest.NewClient(t, srv, "delta-node1",
servertest.WithUser(user1), servertest.WithDeltaUpdates())
servertest.NewClient(t, srv, "delta-node2", servertest.WithUser(user2))
c1.WaitForPeers(t, 1, 15*time.Second)
return srv, c1, findNodeID(t, srv, "delta-node2")
}
assertRemovedAsDelta := func(t *testing.T, c1 *servertest.TestClient, id types.NodeID) {
t.Helper()
assert.EventuallyWithT(t, func(c *assert.CollectT) {
assert.Contains(c, c1.DeltaRemovedPeers(), id.NodeID())
}, 10*time.Second, 50*time.Millisecond, "peer removal never applied as a delta")
}
t.Run("node_deleted", func(t *testing.T) {
t.Parallel()
srv, c1, nodeID2 := setup(t)
node2, ok := srv.State().GetNodeByID(nodeID2)
require.True(t, ok)
changes, err := srv.State().DeleteNode(node2)
require.NoError(t, err)
srv.App.Change(changes...)
assertRemovedAsDelta(t, c1, nodeID2)
})
// The new stream's initial map is a full rebuild, so a removal missed
// while disconnected must still follow as a delta.
t.Run("deleted_while_disconnected", func(t *testing.T) {
t.Parallel()
srv, c1, nodeID2 := setup(t)
node2, ok := srv.State().GetNodeByID(nodeID2)
require.True(t, ok)
c1.Disconnect(t)
changes, err := srv.State().DeleteNode(node2)
require.NoError(t, err)
srv.App.Change(changes...)
c1.Reconnect(t)
assertRemovedAsDelta(t, c1, nodeID2)
})
t.Run("hidden_by_policy", func(t *testing.T) {
t.Parallel()
srv, c1, nodeID2 := setup(t)
changed, err := srv.State().SetPolicy([]byte(`{
"acls": [
{"action": "accept", "src": ["delta-user1@"], "dst": ["delta-user1@:*"]},
{"action": "accept", "src": ["delta-user2@"], "dst": ["delta-user2@:*"]}
]
}`))
require.NoError(t, err)
require.True(t, changed)
changes, err := srv.State().ReloadPolicy()
require.NoError(t, err)
srv.App.Change(changes...)
assertRemovedAsDelta(t, c1, nodeID2)
})
}
func findNodeID(tb testing.TB, srv *servertest.TestServer, hostname string) types.NodeID {
tb.Helper()
+94
View File
@@ -3,6 +3,7 @@ package servertest_test
import (
"context"
"net/netip"
"strings"
"testing"
"time"
@@ -462,3 +463,96 @@ func TestSuggestExitNodeDefaultOnPeerCapMap(t *testing.T) {
viewer.WaitForCondition(t, "peer suggest-exit-node gone after unapprove",
10*time.Second, peerHasCap(false))
}
// firstResolver returns the address of the first DNS resolver in nm.
func firstResolver(nm *netmap.NetworkMap) string {
if nm == nil || len(nm.DNS.Resolvers) == 0 {
return ""
}
return nm.DNS.Resolvers[0].Addr
}
// TestNodeAttrsNextDNS checks a node's DNS config follows each of its
// inputs: the NextDNS profile from nodeAttrs, whether reached through a
// policy reload or a tag change, and the device metadata from its Hostinfo.
// Policy responses do not carry DNSConfig, so each must arrive on its own.
func TestNodeAttrsNextDNS(t *testing.T) {
t.Parallel()
setup := func(t *testing.T) (*servertest.TestServer, *servertest.TestClient) {
t.Helper()
srv := servertest.NewServer(t, servertest.WithDNSResolvers("https://dns.nextdns.io/base"))
user := srv.CreateUser(t, "nd-user")
c := servertest.NewClient(t, srv, "nd-node", servertest.WithUser(user))
c.WaitForCondition(t, "base resolver", 10*time.Second,
func(nm *netmap.NetworkMap) bool {
return strings.HasPrefix(firstResolver(nm), "https://dns.nextdns.io/base?")
})
return srv, c
}
waitResolver := func(t *testing.T, c *servertest.TestClient, prefix string) {
t.Helper()
c.WaitForCondition(t, "resolver "+prefix, 10*time.Second,
func(nm *netmap.NetworkMap) bool {
return strings.HasPrefix(firstResolver(nm), prefix)
})
}
t.Run("policy_reload", func(t *testing.T) {
t.Parallel()
srv, c := setup(t)
reloadPolicy(t, srv, `{
"acls": [{"action": "accept", "src": ["*"], "dst": ["*:*"]}],
"nodeAttrs": [{"target": ["nd-user@"], "attr": ["nextdns:userprof"]}]
}`)
waitResolver(t, c, "https://dns.nextdns.io/userprof?")
})
t.Run("tag_change", func(t *testing.T) {
t.Parallel()
srv, c := setup(t)
reloadPolicy(t, srv, `{
"tagOwners": {"tag:dns": ["nd-user@"]},
"acls": [{"action": "accept", "src": ["*"], "dst": ["*:*"]}],
"nodeAttrs": [{"target": ["tag:dns"], "attr": ["nextdns:tagprof"]}]
}`)
_, tagChange, err := srv.State().SetNodeTags(findNodeID(t, srv, "nd-node"), []string{"tag:dns"})
require.NoError(t, err)
srv.App.Change(tagChange)
waitResolver(t, c, "https://dns.nextdns.io/tagprof?")
})
t.Run("hostname_change", func(t *testing.T) {
t.Parallel()
_, c := setup(t)
c.Direct().SetHostinfo(&tailcfg.Hostinfo{
BackendLogID: "servertest-nd-node",
Hostname: "nd-renamed",
})
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = c.Direct().SendUpdate(ctx)
c.WaitForCondition(t, "renamed device_name", 10*time.Second,
func(nm *netmap.NetworkMap) bool {
return strings.Contains(firstResolver(nm), "device_name=nd-renamed")
})
})
}
+1 -1
View File
@@ -197,7 +197,7 @@ func TestRaceConcurrentServerMutations(t *testing.T) {
wg.Go(func() {
delChange, err := srv.State().DeleteNode(nv2)
if err == nil {
srv.App.Change(delChange)
srv.App.Change(delChange...)
}
})
+29
View File
@@ -16,6 +16,7 @@ import (
"github.com/juanfont/headscale/hscontrol/types"
"tailscale.com/net/memnet"
"tailscale.com/tailcfg"
"tailscale.com/types/dnstype"
)
// TestServer is an in-process Headscale control server suitable for
@@ -43,6 +44,8 @@ type serverConfig struct {
nodeExpiry time.Duration
batcherWorkers int
taildropEnabled bool
magicDNSDomain string
dnsResolvers []string
}
func defaultServerConfig() *serverConfig {
@@ -91,6 +94,18 @@ func WithTaildropEnabled(enabled bool) ServerOption {
return func(c *serverConfig) { c.taildropEnabled = enabled }
}
// WithMagicDNS enables MagicDNS under domain, so map responses carry a
// [tailcfg.DNSConfig].
func WithMagicDNS(domain string) ServerOption {
return func(c *serverConfig) { c.magicDNSDomain = domain }
}
// WithDNSResolvers sets the global DNS resolvers, so map responses carry a
// [tailcfg.DNSConfig].
func WithDNSResolvers(addrs ...string) ServerOption {
return func(c *serverConfig) { c.dnsResolvers = addrs }
}
// NewServer creates and starts a Headscale test server.
// The server is fully functional and accepts real Tailscale control
// protocol connections over Noise.
@@ -137,6 +152,20 @@ func NewServer(tb testing.TB, opts ...ServerOption) *TestServer {
},
}
if sc.magicDNSDomain != "" || len(sc.dnsResolvers) > 0 {
cfg.TailcfgDNSConfig = &tailcfg.DNSConfig{}
}
if sc.magicDNSDomain != "" {
cfg.BaseDomain = sc.magicDNSDomain
cfg.TailcfgDNSConfig.Proxied = true
cfg.TailcfgDNSConfig.Domains = []string{sc.magicDNSDomain}
}
for _, addr := range sc.dnsResolvers {
cfg.TailcfgDNSConfig.Resolvers = append(cfg.TailcfgDNSConfig.Resolvers, &dnstype.Resolver{Addr: addr})
}
app, err := hscontrol.NewHeadscale(&cfg)
if err != nil {
tb.Fatalf("servertest: NewHeadscale: %v", err)
+1 -1
View File
@@ -121,7 +121,7 @@ func TestStressConnectDisconnect(t *testing.T) {
deleteChange, err := srv.State().DeleteNode(nv)
require.NoError(t, err)
srv.App.Change(deleteChange)
srv.App.Change(deleteChange...)
// c2 should see c1 disappear.
c2.WaitForCondition(t, "deleted node gone", 10*time.Second,
+4
View File
@@ -36,6 +36,10 @@ type mapRequestDelta struct {
// changed (see [peerHostinfo]). Only that forces a whole-node resend.
peerHostinfoChanged bool
// dnsMetadataChanged reports whether a Hostinfo field feeding the
// node's NextDNS device metadata (Hostname, OS) changed.
dnsMetadataChanged bool
// routesChanged reports whether announced routes (RoutableIPs)
// changed. Routes are policy and election inputs, so they are tracked
// on their own.
+5 -4
View File
@@ -510,7 +510,7 @@ func TestDeleteNodeKeepsStoreOnDBFailure(t *testing.T) {
c, err := s.DeleteNode(node)
require.NoError(t, s.db.DB.Callback().Delete().Remove("fail_node_delete"))
require.ErrorIs(t, err, errInjectedNodeDelete)
assert.True(t, c.IsEmpty(), "an uncommitted deletion must not stop the node's session")
assert.Empty(t, c, "an uncommitted deletion must not stop the node's session")
_, ok = s.GetNodeByID(nodeID)
assert.True(t, ok, "a database failure must leave the in-memory node available")
@@ -528,11 +528,12 @@ func TestDeleteNodeReturnsRemovalOnPolicyFailure(t *testing.T) {
s.polMan = failingSetNodesPolicyManager{PolicyManager: s.polMan}
c, err := s.DeleteNode(node)
cs, err := s.DeleteNode(node)
require.ErrorIs(t, err, errInjectedPolicyNodeUpdate)
assert.Equal(t, []types.NodeID{nodeID}, c.PeersRemoved,
require.NotEmpty(t, cs)
assert.Equal(t, []types.NodeID{nodeID}, cs[0].PeersRemoved,
"a committed deletion must still notify peers and stop the node's session")
assert.Equal(t, []types.NodeID{nodeID}, c.DeletedNodes,
assert.Equal(t, []types.NodeID{nodeID}, cs[0].DeletedNodes,
"a committed deletion must identify the session to stop")
_, ok = s.GetNodeByID(nodeID)
+50 -27
View File
@@ -191,6 +191,11 @@ type State struct {
// caller snapshot or resurrected by an update racing with deletion.
persistMu sync.Mutex
// selfRefresh holds nodes whose NextDNS device metadata changed, drained
// by [State.DrainSelfRefreshes].
selfRefresh []types.NodeID
selfRefreshMu sync.Mutex
// registerLocks serialises registration per machine key so concurrent
// registrations of the same machine resolve to a single node instead of
// racing the find-then-create section and each creating their own.
@@ -357,21 +362,11 @@ func (s *State) ReloadPolicy() ([]change.Change, error) {
// policies to not propagate correctly when switching between policy types.
s.nodeStore.RebuildPeerMaps()
// Nodes whose CapMap shifted get their self refresh from
// [State.DrainSelfRefreshes] when these changes are dispatched.
//nolint:prealloc // cs starts with one element and may grow
cs := []change.Change{change.PolicyChange()}
// Per-node selective self refresh for nodeAttrs. A broadcast
// [change.PolicyChange] re-renders peer lists and packet filters
// but never repopulates a node's own [tailcfg.Node.CapMap]; that
// lives on the self entry only. The drain returns every node ID
// whose cap output shifted across recent updateLocked calls —
// refreshNodeAttrsLocked appends rather than overwrites so a
// concurrent SetUsers/SetNodes between SetPolicy and the drain
// cannot silently lose the policy-reload diff.
for _, id := range s.polMan.NodesWithChangedCapMap() {
cs = append(cs, change.SelfUpdate(id))
}
// Always call autoApproveNodes during policy reload, regardless of whether
// the policy content has changed. This ensures that routes are re-evaluated
// when they might have been manually disabled but could now be auto-approved
@@ -604,18 +599,18 @@ func (s *State) SaveNode(node types.NodeView) (types.NodeView, change.Change, er
}
// DeleteNode permanently removes a node and cleans up associated resources.
// Once the database deletion commits, the returned change always contains the
// node-removal notification, even if a later policy refresh fails. Callers must
// publish a non-empty change before handling the error so live sessions are
// Once the database deletion commits, the returned changes always start with
// the node-removal notification, even if a later policy refresh fails. Callers
// must publish the changes before handling the error so live sessions are
// still torn down after a committed deletion.
func (s *State) DeleteNode(node types.NodeView) (change.Change, error) {
func (s *State) DeleteNode(node types.NodeView) ([]change.Change, error) {
s.persistMu.Lock()
err := s.db.DeleteNode(node.AsStruct())
if err != nil {
s.persistMu.Unlock()
return change.Change{}, err
return nil, err
}
// The database is the durable source of truth. Only remove the in-memory
@@ -626,21 +621,16 @@ func (s *State) DeleteNode(node types.NodeView) (change.Change, error) {
s.ipAlloc.FreeIPs(node.IPs())
c := change.NodeRemoved(node.ID())
// An explicit removal of its own, ahead of the policy refresh, so peers
// learn of the deletion without depending on their sent-peers tracking.
removed := change.NodeRemoved(node.ID())
// Check if policy manager needs updating after node deletion
policyChange, err := s.updatePolicyManagerNodes()
if err != nil {
return c, fmt.Errorf("updating policy manager after node deletion: %w", err)
return []change.Change{removed, policyChange}, fmt.Errorf("updating policy manager after node deletion: %w", err)
}
if !policyChange.IsEmpty() {
// Merge policy change with NodeRemoved to preserve PeersRemoved info
// This ensures the batcher cleans up the deleted node from its state
c = c.Merge(policyChange)
}
return c, nil
return []change.Change{removed, policyChange}, nil
}
// Connect acquires a control session and returns the resulting changes
@@ -2878,6 +2868,29 @@ func (s *State) UpdatePolicyManagerUsersForTest() error {
return err
}
// DrainSelfRefreshes returns a [change.SelfUpdate] for every node whose own
// entry or DNS config changed since the last call: its policy CapMap, from
// any policy, user or node update, or its NextDNS device metadata. Drained
// where changes are dispatched, so no path has to return the refresh itself.
func (s *State) DrainSelfRefreshes() []change.Change {
ids := s.polMan.NodesWithChangedCapMap()
s.selfRefreshMu.Lock()
ids = append(ids, s.selfRefresh...)
s.selfRefresh = nil
s.selfRefreshMu.Unlock()
slices.Sort(ids)
ids = slices.Compact(ids)
cs := make([]change.Change, 0, len(ids))
for _, id := range ids {
cs = append(cs, change.SelfUpdate(id))
}
return cs
}
// updatePolicyManagerNodes updates the policy manager with current nodes.
// Returns true if the policy changed and notifications should be sent.
// TODO(kradalby): This is a temporary stepping stone, ultimately we should
@@ -3092,6 +3105,10 @@ func (s *State) UpdateNodeFromMapRequest(id types.NodeID, req tailcfg.MapRequest
!hostinfoEqual(currentNode.Hostinfo, newHostinfo)
delta.peerHostinfoChanged = newHostinfo != nil &&
!peerHostinfoEqual(currentNode.Hostinfo, newHostinfo)
delta.dnsMetadataChanged = newHostinfo != nil &&
(currentNode.Hostinfo == nil ||
currentNode.Hostinfo.Hostname != newHostinfo.Hostname ||
currentNode.Hostinfo.OS != newHostinfo.OS)
// A change carrying only an updated LastSeen is not worth a
// full-row database UPDATE plus the O(n) policy rescan
@@ -3193,6 +3210,12 @@ func (s *State) UpdateNodeFromMapRequest(id types.NodeID, req tailcfg.MapRequest
return change.Change{}, fmt.Errorf("%w: %d", ErrNodeNotInNodeStore, id)
}
if delta.dnsMetadataChanged {
s.selfRefreshMu.Lock()
s.selfRefresh = append(s.selfRefresh, id)
s.selfRefreshMu.Unlock()
}
if routeChange {
log.Debug().
Uint64(zf.NodeID, id.Uint64()).
+3
View File
@@ -337,11 +337,14 @@ func FullSelf(nodeID types.NodeID) Change {
}
}
// SelfUpdate sends a node its own entry and the DNS config derived from it
// (NextDNS profile and device metadata).
func SelfUpdate(nodeID types.NodeID) Change {
return Change{
Reason: "self update",
TargetNode: nodeID,
IncludeSelf: true,
IncludeDNS: true,
}
}
+1
View File
@@ -464,6 +464,7 @@ func TestSelfUpdate(t *testing.T) {
assert.Equal(t, "self update", r.Reason)
assert.Equal(t, types.NodeID(42), r.TargetNode)
assert.True(t, r.IncludeSelf)
assert.True(t, r.IncludeDNS, "DNS config derives from the node's own CapMap and Hostinfo")
assert.True(t, r.IsSelfOnly())
}