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

@@ -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.

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:

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