From 13790344ec9dcf06d7cc6096b794193ae111f85a Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Tue, 20 Oct 2026 09:10:13 +0000 Subject: [PATCH] mapper: stop an older initial map removing peers newer streams hold AddNode diffed its map against tracking after sending it, when a newer stream may have recorded its own; it now sends the node a full instead. --- hscontrol/mapper/batcher.go | 83 +++++++++++++++---------- hscontrol/mapper/batcher_test.go | 101 +++++++++++++++++++++++++++++++ hscontrol/mapper/node_conn.go | 6 ++ 3 files changed, 158 insertions(+), 32 deletions(-) diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 53748729b..bf223fda7 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -360,6 +360,9 @@ func (b *Batcher) AddNode( }, ) + // Read before the map is built, so every map recorded after it counts. + tracked := nodeConn.trackedInitialMaps.Load() + // Use the worker pool for controlled concurrency instead of direct generation initialMap, err := b.MapResponseFromChange(id, change.FullSelf(id)) if err != nil { @@ -390,40 +393,56 @@ func (b *Batcher) AddNode( return nil } - // 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], and keep the removed peers tracked across the - // baseline reset below. - // - // 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) + if nodeConn.trackedInitialMaps.Load() != tracked { + // Another stream recorded its map since this one was requested, + // and may hold a newer view: diffing this map against it would + // remove peers that stream rightly holds. Track this stream's + // peers too, and send the node a full to bring every stream to + // the current state. + for _, peer := range initialMap.Peers { + nodeConn.lastSentPeers.Store(peer.ID, struct{}{}) + } + + nodeConn.collapsePendingToFull(nil) + } else { + // 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], and keep the removed peers tracked across the + // baseline reset below. + // + // 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) + } + + removedIDs := nodeConn.computePeerDiff(peerIDs) + + removed := make([]types.NodeID, 0, len(removedIDs)) + for _, peerID := range removedIDs { + //nolint:gosec // NodeID types are equivalent + removed = append(removed, types.NodeID(peerID)) + } + + if len(removed) > 0 { + rm := change.PeersRemoved(removed...) + rm.TargetNode = id + b.AddWork(rm) + } + + nodeConn.updateSentPeers(initialMap) + + // The client has not had these removals as a delta yet. Pending + // may lose the queued one (a full supersedes it, or the node + // leaves before it is sent), so keep them tracked until delivery + // drops them: the next full or reconnect then re-derives them. + for _, peerID := range removedIDs { + nodeConn.lastSentPeers.Store(peerID, struct{}{}) + } } - removedIDs := nodeConn.computePeerDiff(peerIDs) - - removed := make([]types.NodeID, 0, len(removedIDs)) - for _, peerID := range removedIDs { - 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) - - // The client has not had these removals as a delta yet. Pending - // may lose the queued one (a full supersedes it, or the node - // leaves before it is sent), so keep them tracked until delivery - // drops them: the next full or reconnect then re-derives them. - for _, peerID := range removedIDs { - nodeConn.lastSentPeers.Store(peerID, struct{}{}) - } + nodeConn.trackedInitialMaps.Add(1) // Under workMu, so it lands before any newer stream's AddNode // finishes and that stream can leave. diff --git a/hscontrol/mapper/batcher_test.go b/hscontrol/mapper/batcher_test.go index 7f87bed3f..18384f66c 100644 --- a/hscontrol/mapper/batcher_test.go +++ b/hscontrol/mapper/batcher_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/netip" "runtime" + "slices" "strings" "sync" "sync/atomic" @@ -2359,6 +2360,106 @@ func TestEvictedSessionLeavesPeerTrackingAlone(t *testing.T) { assert.True(t, sent, "the peer stays tracked as sent") } +// TestOlderInitialMapLeavesNewerStreamsPeers keeps a stream whose initial map, +// generated while a peer was hidden, in flight until newer streams of the node +// have recorded theirs, which list the peer. Finishing the older call must not +// remove the peer from them, and must bring the older stream up to date. +func TestOlderInitialMapLeavesNewerStreamsPeers(t *testing.T) { + td, cleanup := setupBatcherWithTestData( + t, + NewBatcherAndMapper, + 1, + 2, + normalBufferSize, + ) + defer cleanup() + + b := unwrapBatcher(td.Batcher) + peer, target := td.Nodes[0].n.ID.NodeID(), td.Nodes[1].n.ID + capVer := tailcfg.CapabilityVersion(100) + + peerIDs := func(resp *tailcfg.MapResponse) []tailcfg.NodeID { + ids := make([]tailcfg.NodeID, 0, len(resp.Peers)) + for _, p := range resp.Peers { + ids = append(ids, p.ID) + } + + return ids + } + + _, err := td.State.SetPolicy([]byte(`{"acls":[]}`)) + require.NoError(t, err) + + // Unbuffered and unread: AddNode parks on the initial map send. + olderCh := make(chan *tailcfg.MapResponse) + addErr := make(chan error, 1) + + go func() { addErr <- b.AddNode(target, olderCh, capVer, nil) }() + + waitInitialMapSendParked(t) + + _, err = td.State.SetPolicy( + []byte(`{"acls":[{"action":"accept","src":["*"],"dst":["*:*"]}]}`), + ) + require.NoError(t, err) + + // One short of the cap, so the older stream stays. + newer := make([]chan *tailcfg.MapResponse, maxConcurrentMapSessionsPerNode-1) + for i := range newer { + newer[i] = make(chan *tailcfg.MapResponse, normalBufferSize) + require.NoError(t, b.AddNode(target, newer[i], capVer, nil)) + require.Contains(t, peerIDs(<-newer[i]), peer, "stream %d initial map", i) + } + + // Read from here on, so later sends find the older stream draining. + older := make(chan *tailcfg.MapResponse, normalBufferSize) + + go func() { + for { + select { + case resp := <-olderCh: + older <- resp + case <-t.Context().Done(): + return + } + } + }() + + stale := <-older + require.NotContains(t, peerIDs(stale), peer, "the map predates the policy change") + require.NoError(t, <-addErr) + + // Queued behind anything the older call queued. + b.AddWork(change.DERPMap()) + + for i, ch := range newer { + for done := false; !done; { + select { + case resp := <-ch: + require.NotContains(t, resp.PeersRemoved, peer, "stream %d", i) + done = resp.DERPMap != nil + case <-time.After(5 * time.Second): + t.Fatalf("stream %d never got the DERP map", i) + } + } + } + + nc, ok := b.nodes.Load(target) + require.True(t, ok) + + _, sent := nc.lastSentPeers.Load(peer) + assert.True(t, sent, "the peer stays tracked as sent") + + for listed := false; !listed; { + select { + case resp := <-older: + listed = slices.Contains(peerIDs(resp), peer) + case <-time.After(5 * time.Second): + t.Fatal("the older stream never got the peer") + } + } +} + // TestNodeDeletedWhileChangesPending reproduces issue #2924 where deleting a node // from state while there are pending changes for that node in the batcher causes // "node not found" errors. The race condition occurs when: diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index 7009823a2..e5aebdf77 100644 --- a/hscontrol/mapper/node_conn.go +++ b/hscontrol/mapper/node_conn.go @@ -178,6 +178,12 @@ type multiChannelNodeConn struct { // Partial additions remain removal candidates when retries render newer state. partialPeers *xsync.Map[tailcfg.NodeID, struct{}] + + // trackedInitialMaps counts initial maps recorded in lastSentPeers. + // [Batcher.AddNode] records its map after sending it, outside the workMu + // hold that built it, so a change across that gap means another stream's + // map, possibly built later, was recorded in between. + trackedInitialMaps atomic.Uint64 } // connIDCounter is a monotonically increasing counter used to generate