diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 66b7b582..005dca8a 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -630,6 +630,9 @@ func (b *Batcher) processBatchedChanges() { return true } + // One policy recompute rebuilds the whole netmap; drop same-tick repeats. + pending = change.DedupePolicyChanges(pending) + // Queue a single work item containing all pending changes. // One item per node ensures a single worker processes them // sequentially, preventing out-of-order delivery. diff --git a/hscontrol/mapper/batcher_test.go b/hscontrol/mapper/batcher_test.go index 738c0b6a..49aeb54d 100644 --- a/hscontrol/mapper/batcher_test.go +++ b/hscontrol/mapper/batcher_test.go @@ -1104,6 +1104,72 @@ func TestBatcherWorkQueueBatching(t *testing.T) { } } +// TestBatcherCoalescesPolicyRecomputesPerTick proves that many identical +// broadcast policy changes arriving in a single batcher tick collapse to one +// runtime peer recompute per node. Without coalescing, each PolicyChange drives +// a separate full netmap rebuild for every connected node (the reconnect-storm +// fan-out); with it, a node sees at most one recompute per tick. +func TestBatcherCoalescesPolicyRecomputesPerTick(t *testing.T) { + for _, bf := range allBatcherFunctions { + t.Run(bf.name, func(t *testing.T) { + const ( + nodesPerUser = 4 + policyChangesPerTick = 8 + ) + + testData, cleanup := setupBatcherWithTestData(t, bf.fn, 1, nodesPerUser, 100) + defer cleanup() + + batcher := testData.Batcher + for i := range testData.Nodes { + n := &testData.Nodes[i] + require.NoError(t, batcher.AddNode(n.n.ID, n.ch, tailcfg.CapabilityVersion(100), nil)) + } + + // Many identical broadcast policy changes, then a DERP-map change as + // a sentinel. All land in one tick; the sentinel rides the same work + // item after the recompute(s), so its arrival marks the end of this + // tick's policy responses for a node. + for range policyChangesPerTick { + batcher.AddWork(change.PolicyChange()) + } + + batcher.AddWork(change.DERPMap()) + + for i := range testData.Nodes { + id := testData.Nodes[i].n.ID + ch := testData.Nodes[i].ch + + policyResponses := 0 + deadline := time.After(2 * time.Second) + + drain: + for { + select { + case resp := <-ch: + switch { + case resp.DERPMap != nil && len(resp.Peers) == 0: + // Sentinel: every policy recompute for this tick has + // already been delivered to this node. + break drain + case len(resp.PacketFilters) > 0 && len(resp.Peers) == 0: + // A runtime peer recompute (policyChangeResponse): + // packet filters and incremental peers, no full list. + policyResponses++ + } + case <-deadline: + t.Fatalf("node %d never received the DERP sentinel", id) + } + } + + assert.LessOrEqualf(t, policyResponses, 1, + "node %d received %d policy recomputes in one tick; identical recomputes must coalesce to one", + id, policyResponses) + } + }) + } +} + // TestBatcherWorkerChannelSafety tests that worker goroutines handle closed // channels safely without panicking when processing work items. // diff --git a/hscontrol/types/change/change.go b/hscontrol/types/change/change.go index abfcaee2..5b3c933f 100644 --- a/hscontrol/types/change/change.go +++ b/hscontrol/types/change/change.go @@ -250,6 +250,41 @@ func FilterForNode(nodeID types.NodeID, rs []Change) []Change { return result } +// IsBroadcastPolicyChange reports whether r is a tailnet-wide policy recompute +// with no per-node payload. A recompute reads the current snapshot, so every +// such change is interchangeable and same-tick duplicates are redundant. A +// targeted or self-update ([Change.OriginNode]) recompute is per-node, so it is +// not one of these. +func (r Change) IsBroadcastPolicyChange() bool { + return r.RequiresRuntimePeerComputation && !r.IsTargetedToNode() && r.OriginNode == 0 +} + +// DedupePolicyChanges keeps the first broadcast policy change in a tick and +// drops the rest: each rebuilds a node's whole netmap from the same snapshot, so +// the repeats are wasted work. Order and all other changes are preserved. +func DedupePolicyChanges(changes []Change) []Change { + if len(changes) < 2 { + return changes + } + + out := make([]Change, 0, len(changes)) + seen := false + + for _, r := range changes { + if r.IsBroadcastPolicyChange() { + if seen { + continue + } + + seen = true + } + + out = append(out, r) + } + + return out +} + func uniqueNodeIDs(ids []types.NodeID) []types.NodeID { if len(ids) == 0 { return nil diff --git a/hscontrol/types/change/change_test.go b/hscontrol/types/change/change_test.go index 729dc468..c5086267 100644 --- a/hscontrol/types/change/change_test.go +++ b/hscontrol/types/change/change_test.go @@ -296,6 +296,99 @@ func TestChange_Merge(t *testing.T) { } } +func TestChange_IsBroadcastPolicyChange(t *testing.T) { + originUpdate := PolicyChange() + originUpdate.OriginNode = 7 + + targeted := PolicyChange() + targeted.TargetNode = 7 + + tests := []struct { + name string + c Change + want bool + }{ + {name: "policy change", c: PolicyChange(), want: true}, + {name: "self-update recompute", c: originUpdate, want: false}, + {name: "targeted recompute", c: targeted, want: false}, + {name: "online patch", c: NodeOnline(1), want: false}, + {name: "full update", c: FullUpdate(), want: false}, + {name: "derp map", c: DERPMap(), want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, tt.c.IsBroadcastPolicyChange()) + }) + } +} + +func TestDedupePolicyChanges(t *testing.T) { + // originRecompute is a runtime recompute carrying node-specific payload + // (OriginNode), so it is not the canonical broadcast PolicyChange and must + // never be coalesced away. + originRecompute := PolicyChange() + originRecompute.OriginNode = 7 + + tests := []struct { + name string + changes []Change + want []Change + }{ + { + name: "nil is a no-op", + changes: nil, + want: nil, + }, + { + name: "single policy change is unchanged", + changes: []Change{PolicyChange()}, + want: []Change{PolicyChange()}, + }, + { + name: "identical policy changes collapse to one", + changes: []Change{PolicyChange(), PolicyChange(), PolicyChange()}, + want: []Change{PolicyChange()}, + }, + { + name: "peer patches survive between collapsed policy changes", + changes: []Change{ + NodeOnline(1), PolicyChange(), NodeOnline(2), PolicyChange(), NodeOffline(3), + }, + want: []Change{ + NodeOnline(1), PolicyChange(), NodeOnline(2), NodeOffline(3), + }, + }, + { + name: "NodeAdded is preserved, not treated as a recompute", + changes: []Change{PolicyChange(), NodeAdded(5), PolicyChange()}, + want: []Change{PolicyChange(), NodeAdded(5)}, + }, + { + name: "recompute carrying OriginNode is kept alongside the canonical one", + changes: []Change{PolicyChange(), originRecompute, PolicyChange()}, + want: []Change{PolicyChange(), originRecompute}, + }, + { + name: "non-canonical recomputes are not collapsed", + changes: []Change{originRecompute, originRecompute}, + want: []Change{originRecompute, originRecompute}, + }, + { + name: "changes without any recompute are unchanged", + changes: []Change{NodeOnline(1), DERPMap()}, + want: []Change{NodeOnline(1), DERPMap()}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := DedupePolicyChanges(tt.changes) + assert.Equal(t, tt.want, got) + }) + } +} + func TestChange_Constructors(t *testing.T) { tests := []struct { name string