diff --git a/hscontrol/mapper/batch_coalesce_test.go b/hscontrol/mapper/batch_coalesce_test.go index bcbae014b..918bafcc7 100644 --- a/hscontrol/mapper/batch_coalesce_test.go +++ b/hscontrol/mapper/batch_coalesce_test.go @@ -73,9 +73,9 @@ func TestProcessBatchedChangesSkipsNodesWithoutConnection(t *testing.T) { // 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. +// when the workers fall behind. Below the cap, 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) @@ -83,11 +83,11 @@ func TestAppendPendingStaysBoundedWhileInFlight(t *testing.T) { var all []change.Change - for i := range 10000 { - peer := types.NodeID(i%500 + 2) //nolint:gosec // test + for i := range 3000 { + peer := types.NodeID(i%50 + 2) //nolint:gosec // test cs := []change.Change{change.NodeAdded(peer), change.NodeOnline(peer), change.PolicyChange()} - if i%7 == 0 { + if i%100 == 0 { cs = append(cs, change.NodeRemoved(types.NodeID(i+1000))) //nolint:gosec // test } @@ -103,9 +103,36 @@ func TestAppendPendingStaysBoundedWhileInFlight(t *testing.T) { nc.pendingMu.Unlock() assert.Less(t, len(pending), 100, "pending must not grow with every change") + assert.False(t, pending[0].IsFull(), "test sanity: the backlog stays below the cap") coalesce := func(cs []change.Change) []change.Change { return change.CoalescePeerDeltas(id, change.DedupePolicyChanges(cs)) } assert.Equal(t, coalesce(all), coalesce(pending)) } + +// TestAppendPendingBoundsFoldedPatches queues patch-only updates for one peer +// while a node's bundle stays in flight. Folding keeps each patch, so it must +// count them: a backlog folded into one change still grows by every patch, +// and only a collapse to a full map bounds it. +func TestAppendPendingBoundsFoldedPatches(t *testing.T) { + id := types.NodeID(1) + nc := newMultiChannelNodeConn(id, nil) + nc.inFlight.Store(true) + + for i := range 10000 { + if i%2 == 0 { + nc.appendPending(change.NodeOnline(2)) + } else { + nc.appendPending(change.NodeOffline(2)) + } + } + + nc.pendingMu.Lock() + pending := slices.Clone(nc.pending) + nc.pendingMu.Unlock() + + require.NotEmpty(t, pending) + assert.True(t, pending[0].IsFull(), "a backlog past the cap must collapse to a full map") + assert.LessOrEqual(t, countPatches(pending), maxPendingSize, "pending must not keep every patch") +} diff --git a/hscontrol/mapper/batcher_unit_test.go b/hscontrol/mapper/batcher_unit_test.go index 0caa7fbdc..c02f9c4a6 100644 --- a/hscontrol/mapper/batcher_unit_test.go +++ b/hscontrol/mapper/batcher_unit_test.go @@ -449,6 +449,7 @@ func TestPendingBacklogCollapsesToCurrentState(t *testing.T) { pending := nc.drainPending() require.LessOrEqual(t, len(pending), 256, "state backlog must collapse while preserving pings") + require.True(t, pending[0].IsFull(), "the cap must collapse the backlog by itself") // A retry must also collapse against a newer full update and keep older pings first. nc.collapsePendingToFull([]change.Change{change.PingNode(id, &tailcfg.PingRequest{URL: "https://new"})}) diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index 058d70238..8fb15f716 100644 --- a/hscontrol/mapper/node_conn.go +++ b/hscontrol/mapper/node_conn.go @@ -19,7 +19,7 @@ import ( "tailscale.com/tailcfg" ) -// minPendingFold is the pending length below which appendPending leaves the +// minPendingFold is the pending size below which appendPending leaves the // folding to the drain. const minPendingFold = 64 @@ -45,8 +45,9 @@ var errConnectionBackpressure = errors.New("connection queue full") // Match the initial-map timeout while staying below reconnect grace and client keepalive. const connectionStallTimeout = 5 * time.Second -// A full snapshot supersedes state backlog; pings remain one-shot work. -const maxPendingChanges = 256 +// A full snapshot supersedes state backlog past this size; pings remain +// one-shot work. +const maxPendingSize = 256 // connectionEntry represents a single connection to a node. type connectionEntry struct { @@ -122,7 +123,11 @@ type multiChannelNodeConn struct { pendingMu sync.Mutex pending []change.Change - // pendingFoldAt is the pending length at which appendPending folds the + // pendingSize is [changesSize] of pending. A fold keeps every peer and + // patch it merges, so the change count alone hides the backlog. + pendingSize int + + // pendingFoldAt is the pending size at which appendPending folds the // backlog again: twice what the last fold left, so appends stay // amortized O(1). pendingFoldAt int @@ -315,19 +320,20 @@ func (mc *multiChannelNodeConn) appendPending(changes ...change.Change) { mc.pendingMu.Lock() mc.pending = append(mc.pending, changes...) + mc.pendingSize += changesSize(changes) fullAtHead := len(mc.pending) > 0 && mc.pending[0].IsFull() // 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. Folding first leaves the // cap's collapse to a full map for backlogs that do not fold. - if !fullAtHead && len(mc.pending) >= max(mc.pendingFoldAt, minPendingFold) { - mc.pending = change.CoalescePeerDeltas(mc.id, change.DedupePolicyChanges(mc.pending)) - mc.pendingFoldAt = 2 * len(mc.pending) + if !fullAtHead && mc.pendingSize >= max(mc.pendingFoldAt, minPendingFold) { + mc.setPendingLocked(change.CoalescePeerDeltas(mc.id, change.DedupePolicyChanges(mc.pending))) + mc.pendingFoldAt = 2 * mc.pendingSize } - if fullAtHead || len(mc.pending) > maxPendingChanges { - mc.pending = change.CollapseToFull(mc.id, mc.pending) + if fullAtHead || mc.pendingSize > maxPendingSize { + mc.setPendingLocked(change.CollapseToFull(mc.id, mc.pending)) } mc.pendingMu.Unlock() } @@ -343,8 +349,10 @@ func (mc *multiChannelNodeConn) prependPending(changes ...change.Change) { (len(changes) > 0 && changes[0].IsFull()) mc.pending = append(changes, mc.pending...) - if len(mc.pending) > maxPendingChanges || full { - mc.pending = change.CollapseToFull(mc.id, mc.pending) + mc.pendingSize += changesSize(changes) + + if full || mc.pendingSize > maxPendingSize { + mc.setPendingLocked(change.CollapseToFull(mc.id, mc.pending)) } mc.pendingMu.Unlock() } @@ -353,7 +361,7 @@ func (mc *multiChannelNodeConn) prependPending(changes ...change.Change) { // 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.setPendingLocked(change.CollapseToFull(mc.id, slices.Concat(mc.pending, incoming))) mc.pendingMu.Unlock() } @@ -362,12 +370,29 @@ func (mc *multiChannelNodeConn) collapsePendingToFull(incoming []change.Change) func (mc *multiChannelNodeConn) drainPending() []change.Change { mc.pendingMu.Lock() p := mc.pending - mc.pending = nil + mc.setPendingLocked(nil) mc.pendingMu.Unlock() return p } +// setPendingLocked replaces pending. Caller must hold pendingMu. +func (mc *multiChannelNodeConn) setPendingLocked(p []change.Change) { + mc.pending = p + mc.pendingSize = changesSize(p) +} + +// changesSize is how much pending holds for changes: each change, and the +// peers and patches it carries. +func changesSize(changes []change.Change) int { + n := len(changes) + for _, c := range changes { + n += len(c.PeersChanged) + len(c.PeersRemoved) + len(c.PeerPatches) + } + + return n +} + // takePending drains the pending changes for rendering. A node without a // connection gets none and its changes are dropped: send would discard what // they render, and its next connection opens with a full map rendered after