From 8a7921ae45ae0debec3d2e2900535c8238e06230 Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Thu, 22 Oct 2026 03:37:49 +0000 Subject: [PATCH] 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 | 40 +++++++++++++++++++ hscontrol/mapper/batcher_concurrency_test.go | 41 +++++++++++--------- hscontrol/mapper/node_conn.go | 17 ++++++++ 3 files changed, 79 insertions(+), 19 deletions(-) diff --git a/hscontrol/mapper/batch_coalesce_test.go b/hscontrol/mapper/batch_coalesce_test.go index bd7ae483d..bcbae014b 100644 --- a/hscontrol/mapper/batch_coalesce_test.go +++ b/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)) +} diff --git a/hscontrol/mapper/batcher_concurrency_test.go b/hscontrol/mapper/batcher_concurrency_test.go index 4acaab22c..3bdd92c43 100644 --- a/hscontrol/mapper/batcher_concurrency_test.go +++ b/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 - // data loss: typically 30-50% of changes are silently dropped. - pending := getPendingForNode(lb.b, targetNode) - t.Logf("targeted changes: expected=%d, got=%d (lost=%d)", - goroutines, len(pending), goroutines-len(pending)) + // All 100 patches MUST be present; pending may fold them into fewer + // changes. The Load→append→Store race causes data loss: typically + // 30-50% of changes are silently dropped. + got := countPatches(getPendingForNode(lb.b, targetNode)) + t.Logf("targeted patches: expected=%d, got=%d (lost=%d)", + goroutines, got, goroutines-got) - assert.Len(t, pending, goroutines, - "addToBatch lost %d/%d targeted changes under concurrent access", - goroutines-len(pending), goroutines) + assert.Equal(t, goroutines, got, + "addToBatch lost %d/%d targeted patches under concurrent access", + 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()) diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index cba71ce5f..fe38c3978 100644 --- a/hscontrol/mapper/node_conn.go +++ b/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() }