mapper: fold a node's pending backlog while its bundle is in flight

Behind workers, pending grew by every change until the next drain; under
ci-churn that held ~900 MB of queued changes.

hscontrol/mapper/batch_coalesce_test.go

@@ -1,6 +1,7 @@
 package mapper
 
 import (
+	"slices"
 	"testing"
 	"time"
 
@@ -69,3 +70,42 @@ func TestProcessBatchedChangesSkipsNodesWithoutConnection(t *testing.T) {
 	assert.Empty(t, b.workCh, "no work may be queued for a node without a connection")
 	assert.False(t, nc.inFlight.Load())
 }
+
+// TestAppendPendingStaysBoundedWhileInFlight appends a churn of joins,
+// deletions and policy changes to a node whose bundle stays in flight, as
+// when the workers fall behind. Pending must stay as small as its coalesced
+// form instead of growing by one entry per change, and draining it must
+// render what coalescing the whole backlog at once would.
+func TestAppendPendingStaysBoundedWhileInFlight(t *testing.T) {
+	id := types.NodeID(1)
+	nc := newMultiChannelNodeConn(id, nil)
+	nc.inFlight.Store(true)
+
+	var all []change.Change
+
+	for i := range 10000 {
+		peer := types.NodeID(i%500 + 2) //nolint:gosec // test
+
+		cs := []change.Change{change.NodeAdded(peer), change.NodeOnline(peer), change.PolicyChange()}
+		if i%7 == 0 {
+			cs = append(cs, change.NodeRemoved(types.NodeID(i+1000))) //nolint:gosec // test
+		}
+
+		for _, c := range cs {
+			nc.appendPending(c)
+		}
+
+		all = append(all, cs...)
+	}
+
+	nc.pendingMu.Lock()
+	pending := slices.Clone(nc.pending)
+	nc.pendingMu.Unlock()
+
+	assert.Less(t, len(pending), 100, "pending must not grow with every change")
+
+	coalesce := func(cs []change.Change) []change.Change {
+		return change.CoalescePeerDeltas(id, change.DedupePolicyChanges(cs))
+	}
+	assert.Equal(t, coalesce(all), coalesce(pending))
+}

hscontrol/mapper/batcher_concurrency_test.go

@@ -213,15 +213,27 @@ func TestAddToBatch_ConcurrentTargeted_NoDataLoss(t *testing.T) {
 
 	require.Zero(t, panics, "no panics expected")
 
-	// All 100 changes MUST be present. The Load→append→Store race causes
+	// All 100 patches MUST be present; pending may fold them into fewer
-	// data loss: typically 30-50% of changes are silently dropped.
+	// changes. The Load→append→Store race causes data loss: typically
-	pending := getPendingForNode(lb.b, targetNode)
+	// 30-50% of changes are silently dropped.
-	t.Logf("targeted changes: expected=%d, got=%d (lost=%d)",
+	got := countPatches(getPendingForNode(lb.b, targetNode))
-		goroutines, len(pending), goroutines-len(pending))
+	t.Logf("targeted patches: expected=%d, got=%d (lost=%d)",
+		goroutines, got, goroutines-got)
 
-	assert.Len(t, pending, goroutines,
+	assert.Equal(t, goroutines, got,
-		"addToBatch lost %d/%d targeted changes under concurrent access",
+		"addToBatch lost %d/%d targeted patches under concurrent access",
-		goroutines-len(pending), goroutines)
+		goroutines-got, goroutines)
+}
+
+// countPatches counts the peer patches in changes, which pending folds into
+// fewer changes without dropping any.
+func countPatches(cs []change.Change) int {
+	n := 0
+	for _, c := range cs {
+		n += len(c.PeerPatches)
+	}
+
+	return n
 }
 
 // TestAddToBatch_ConcurrentBroadcast verifies that concurrent broadcasts
@@ -591,24 +603,15 @@ func TestProcessBatchedChanges_ConcurrentAdd_NoDataLoss(t *testing.T) {
 	// Count the patches across all bundled work items in the channel. Each
 	// work item bundles a node's pending changes, coalescing peer deltas, so
 	// the patches, not the changes, are what must all arrive.
-	patches := func(cs []change.Change) int {
-		n := 0
-		for _, c := range cs {
-			n += len(c.PeerPatches)
-		}
-
-		return n
-	}
-
 	queuedChanges := 0
 
 	workItems := len(lb.b.workCh)
 	for range workItems {
 		w := <-lb.b.workCh
-		queuedChanges += patches(w.changes)
+		queuedChanges += countPatches(w.changes)
 	}
 	// Also count any still-pending
-	remaining := patches(getPendingForNode(lb.b, types.NodeID(1)))
+	remaining := countPatches(getPendingForNode(lb.b, types.NodeID(1)))
 
 	total := queuedChanges + remaining
 	added := int(addedCount.Load())

hscontrol/mapper/node_conn.go

@@ -19,6 +19,10 @@ import (
 	"tailscale.com/tailcfg"
 )
 
+// minPendingFold is the pending length below which appendPending leaves the
+// folding to the drain.
+const minPendingFold = 64
+
 // errNoActiveConnections is returned by [multiChannelNodeConn.send] when a node
 // has no active connections (disconnected but kept in the batcher for rapid
 // reconnection). Callers must not update peer tracking state (lastSentPeers)
@@ -105,6 +109,11 @@ type multiChannelNodeConn struct {
 	pendingMu sync.Mutex
 	pending   []change.Change
 
+	// pendingFoldAt is the pending length at which appendPending folds the
+	// backlog again: twice what the last fold left, so appends stay
+	// amortized O(1).
+	pendingFoldAt int
+
 	// workMu serializes change processing for this node across batch ticks.
 	// Without this, two workers could process consecutive ticks' bundles
 	// concurrently, causing out-of-order [tailcfg.MapResponse] delivery and races
@@ -286,6 +295,14 @@ func (mc *multiChannelNodeConn) offlineDuration() time.Duration {
 func (mc *multiChannelNodeConn) appendPending(changes ...change.Change) {
 	mc.pendingMu.Lock()
 	mc.pending = append(mc.pending, changes...)
+
+	// Pending drains only once the bundle in flight is delivered, so behind
+	// workers it grew by every change; it renders folded anyway, and
+	// folding twice gives what folding once does.
+	if len(mc.pending) >= max(mc.pendingFoldAt, minPendingFold) {
+		mc.pending = change.CoalescePeerDeltas(mc.id, change.DedupePolicyChanges(mc.pending))
+		mc.pendingFoldAt = 2 * len(mc.pending)
+	}
 	mc.pendingMu.Unlock()
 }