mapper: preserve node pings when a full update is queued

Updates #3523

(cherry picked from commit 17cb110892)
(cherry picked from commit 7c2260c602)
(cherry picked from commit 6875c4bc8, test only)

hscontrol/mapper/batcher.go

@@ -709,17 +709,15 @@ func (b *Batcher) addToBatch(changes ...change.Change) {
 		}
 	}
 
-	// Short circuit if any of the changes is a full update, which
-	// means we can skip sending individual changes.
+	// A full update supersedes every state change pending or in this call,
+	// but not the pings addressed to a node; those follow the full.
 	if change.HasFull(changes) {
 		b.nodes.Range(func(_ types.NodeID, nc *multiChannelNodeConn) bool {
 			if nc == nil {
 				return true
 			}
 
-			nc.pendingMu.Lock()
-			nc.pending = []change.Change{change.FullUpdate()}
-			nc.pendingMu.Unlock()
+			nc.collapsePendingToFull(changes)
 
 			return true
 		})

hscontrol/mapper/batcher_concurrency_test.go

@@ -278,6 +278,96 @@ func TestAddToBatch_FullUpdateOverrides(t *testing.T) {
 	})
 }
 
+// TestAddToBatch_FullUpdateKeepsPingRequests verifies that a full update
+// keeps the pings it cannot carry: one pending before the full, and one in
+// the same call. A ping lost here times out its caller.
+func TestAddToBatch_FullUpdateKeepsPingRequests(t *testing.T) {
+	lb := setupLightweightBatcher(t, 3, 10)
+	defer lb.cleanup()
+
+	prPending := &tailcfg.PingRequest{URL: "https://example.com/ping/pending"}
+	prSameCall := &tailcfg.PingRequest{URL: "https://example.com/ping/same-call"}
+
+	lb.b.addToBatch(change.PingNode(1, prPending))
+	lb.b.addToBatch(change.NodeOnline(3))
+	lb.b.addToBatch(change.PingNode(2, prSameCall), change.UserRemoved())
+
+	want := map[types.NodeID][]change.Change{
+		1: {change.FullUpdate(), change.PingNode(1, prPending)},
+		2: {change.FullUpdate(), change.PingNode(2, prSameCall)},
+		3: {change.FullUpdate()},
+	}
+
+	for id, w := range want {
+		assert.Equal(t, w, getPendingForNode(lb.b, id), "node %d", id)
+	}
+
+	// A second full neither stacks nor drops the rescued ping.
+	lb.b.addToBatch(change.FullUpdate())
+
+	for id, w := range want {
+		assert.Equal(t, w, getPendingForNode(lb.b, id), "node %d after second full", id)
+	}
+}
+
+// TestAddToBatch_ConcurrentPingAndFullUpdate_NoPingLoss races pings against
+// fulls: every ping must end up pending exactly once, on its own node, behind
+// a single full.
+func TestAddToBatch_ConcurrentPingAndFullUpdate_NoPingLoss(t *testing.T) {
+	const (
+		nodes = 4
+		pings = 64
+		fulls = 16
+	)
+
+	lb := setupLightweightBatcher(t, nodes, 10)
+	defer lb.cleanup()
+
+	target := func(i int) types.NodeID {
+		return types.NodeID(i%nodes + 1) //nolint:gosec // test with small values
+	}
+
+	panics := runConcurrentlyWithTimeout(t, pings+fulls, 10*time.Second, func(i int) {
+		if i < pings {
+			lb.b.addToBatch(change.PingNode(target(i), &tailcfg.PingRequest{
+				URL: fmt.Sprintf("ping-%d", i),
+			}))
+
+			return
+		}
+
+		lb.b.addToBatch(change.FullUpdate())
+	})
+	require.Zero(t, panics)
+
+	seen := make(map[string]types.NodeID, pings)
+
+	for id := range lb.channels {
+		pending := getPendingForNode(lb.b, id)
+		require.NotEmpty(t, pending, "node %d", id)
+		assert.True(t, pending[0].IsFull(), "node %d: full must lead", id)
+
+		for _, c := range pending[1:] {
+			require.NotNil(t, c.PingRequest, "node %d: only pings follow the full", id)
+			assert.Equal(t, change.PingNode(id, c.PingRequest), c, "node %d: ping-only", id)
+
+			prev, dup := seen[c.PingRequest.URL]
+			assert.False(t, dup, "%s pending on node %d and %d", c.PingRequest.URL, prev, id)
+
+			seen[c.PingRequest.URL] = id
+		}
+	}
+
+	for i := range pings {
+		url := fmt.Sprintf("ping-%d", i)
+		got, ok := seen[url]
+
+		if assert.True(t, ok, "%s lost", url) {
+			assert.Equal(t, target(i), got, "%s on wrong node", url)
+		}
+	}
+}
+
 // TestAddToBatch_NodeRemovalCleanup verifies that a permanent node deletion
 // cleans up the node from the batcher's internal state.
 func TestAddToBatch_NodeRemovalCleanup(t *testing.T) {

hscontrol/mapper/batcher_test.go

@@ -1173,6 +1173,78 @@ func TestBatcherCoalescesPolicyRecomputesPerTick(t *testing.T) {
 	}
 }
 
+// TestBatcherPingSurvivesFullUpdate queues a ping and a full in one AddWork
+// call against real connections: the target renders the full and then a
+// ping-only frame, and no other node sees the ping.
+func TestBatcherPingSurvivesFullUpdate(t *testing.T) {
+	for _, bf := range allBatcherFunctions {
+		t.Run(bf.name, func(t *testing.T) {
+			testData, cleanup := setupBatcherWithTestData(t, bf.fn, 1, 2, normalBufferSize)
+			defer cleanup()
+
+			batcher := testData.Batcher
+			target, other := &testData.Nodes[0], &testData.Nodes[1]
+
+			for _, n := range []*node{target, other} {
+				require.NoError(t, batcher.AddNode(n.n.ID, n.ch, tailcfg.CapabilityVersion(100), nil))
+			}
+
+			// Settle initial maps and online patches so only this call's
+			// frames remain.
+			drainChannelTimeout(target.ch, 300*time.Millisecond)
+			drainChannelTimeout(other.ch, 300*time.Millisecond)
+
+			pr := &tailcfg.PingRequest{URL: "https://example.com/ping", Log: true}
+			batcher.AddWork(change.PingNode(target.n.ID, pr), change.UserRemoved())
+
+			collect := func(ch <-chan *tailcfg.MapResponse) []*tailcfg.MapResponse {
+				var frames []*tailcfg.MapResponse
+
+				deadline := time.After(updateTimeout)
+				quiet := time.NewTimer(time.Hour)
+
+				defer quiet.Stop()
+
+				for {
+					select {
+					case resp := <-ch:
+						frames = append(frames, resp)
+
+						quiet.Reset(300 * time.Millisecond)
+					case <-quiet.C:
+						return frames
+					case <-deadline:
+						return frames
+					}
+				}
+			}
+
+			isFull := func(r *tailcfg.MapResponse) bool {
+				// An unchanged self may be omitted for this connection, while
+				// the full peer list and DERP map still distinguish the full.
+				return r.DERPMap != nil && len(r.Peers) > 0
+			}
+
+			frames := collect(target.ch)
+			require.Len(t, frames, 2, "target: full then ping")
+			assert.True(t, isFull(frames[0]), "target: first frame is the full")
+			assert.Nil(t, frames[0].PingRequest, "target: the full carries no ping")
+
+			ping := frames[1]
+			assert.Equal(t, pr, ping.PingRequest)
+			assert.Nil(t, ping.Node, "ping frame is ping-only")
+			assert.Nil(t, ping.DERPMap, "ping frame is ping-only")
+			assert.Empty(t, ping.Peers, "ping frame is ping-only")
+			assert.Empty(t, ping.PeersChangedPatch, "ping frame is ping-only")
+
+			frames = collect(other.ch)
+			require.Len(t, frames, 1, "other: only the full")
+			assert.True(t, isFull(frames[0]), "other: frame is the full")
+			assert.Nil(t, frames[0].PingRequest, "other: no ping")
+		})
+	}
+}
+
 // TestBatcherWorkerChannelSafety tests that worker goroutines handle closed
 // channels safely without panicking when processing work items.
 //

hscontrol/mapper/node_conn.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"slices"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -291,6 +292,14 @@ func (mc *multiChannelNodeConn) prependPending(changes ...change.Change) {
 	mc.pendingMu.Unlock()
 }
 
+// collapsePendingToFull replaces pending, together with incoming, by a
+// single full update and the pings it cannot carry ([change.CollapseToFull]).
+func (mc *multiChannelNodeConn) collapsePendingToFull(incoming []change.Change) {
+	mc.pendingMu.Lock()
+	mc.pending = change.CollapseToFull(mc.id, slices.Concat(mc.pending, incoming))
+	mc.pendingMu.Unlock()
+}
+
 // drainPending atomically removes and returns all pending changes.
 // Returns nil if there are no pending changes.
 func (mc *multiChannelNodeConn) drainPending() []change.Change {