mapper, change: coalesce duplicate policy recomputes per tick

Deduplicate broadcast policy changes within a tick so peers don't recompute
their netmap more than needed during a reconnect storm.

Updates #3293
This commit is contained in:
Kristoffer Dalby
2026-06-03 13:21:59 +00:00
parent 7706552c99
commit cffdb77c8b
4 changed files with 197 additions and 0 deletions
+3
View File
@@ -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.
+66
View File
@@ -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.
//
+35
View File
@@ -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
+93
View File
@@ -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