diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index ad63c42c1..325936836 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -750,8 +750,10 @@ func (b *Batcher) processBatchedChanges() { return true } - // One policy recompute rebuilds the whole netmap; drop same-tick repeats. - pending = change.DedupePolicyChanges(pending) + // One policy recompute rebuilds the whole netmap; drop same-tick + // repeats. A frame per peer delta lets a join storm outgrow the + // session channel within one tick, so render each run as one. + pending = change.CoalescePeerDeltas(nodeID, change.DedupePolicyChanges(pending)) // Queue a single work item containing all pending changes. nc.inFlight.Store(true) diff --git a/hscontrol/mapper/batcher_concurrency_test.go b/hscontrol/mapper/batcher_concurrency_test.go index 90c60c8f2..4acaab22c 100644 --- a/hscontrol/mapper/batcher_concurrency_test.go +++ b/hscontrol/mapper/batcher_concurrency_test.go @@ -588,18 +588,27 @@ func TestProcessBatchedChanges_ConcurrentAdd_NoDataLoss(t *testing.T) { // One final process to flush any remaining lb.b.processBatchedChanges() - // Count total changes across all bundled work items in the channel. - // Each work item may contain multiple changes since processBatchedChanges - // bundles all pending changes per node into a single work item. + // 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 += len(w.changes) + queuedChanges += patches(w.changes) } // Also count any still-pending - remaining := len(getPendingForNode(lb.b, types.NodeID(1))) + remaining := patches(getPendingForNode(lb.b, types.NodeID(1))) total := queuedChanges + remaining added := int(addedCount.Load()) diff --git a/hscontrol/mapper/batcher_test.go b/hscontrol/mapper/batcher_test.go index 798b1e34a..3145e05f8 100644 --- a/hscontrol/mapper/batcher_test.go +++ b/hscontrol/mapper/batcher_test.go @@ -1173,6 +1173,55 @@ func TestBatcherCoalescesPolicyRecomputesPerTick(t *testing.T) { } } +// TestBatcherCoalescesPeerDeltasPerTick lands a join storm, every other node +// added and coming online, in one tick. Rendered one frame per change, a +// storm outgrows the session channel within a single tick and the batcher +// drops the connection as stale; the run must reach the node as one frame. +func TestBatcherCoalescesPeerDeltasPerTick(t *testing.T) { + for _, bf := range allBatcherFunctions { + t.Run(bf.name, func(t *testing.T) { + testData, cleanup := setupBatcherWithTestData(t, bf.fn, 1, 8, 100) + defer cleanup() + + recv := &testData.Nodes[0] + require.NoError(t, testData.Batcher.Batcher.AddNode(recv.n.ID, recv.ch, tailcfg.CapabilityVersion(100), nil)) + + initial := <-recv.ch + require.NotNil(t, initial.Node, "test sanity: initial map first") + + var storm []change.Change + + for i := 1; i < len(testData.Nodes); i++ { + id := testData.Nodes[i].n.ID + storm = append(storm, change.NodeAdded(id), change.NodeOnline(id)) + } + + // One AddWork lands in one tick; the DERP map change ends it. + testData.Batcher.AddWork(append(storm, change.DERPMap())...) + + var frames []*tailcfg.MapResponse + + deadline := time.After(5 * time.Second) + + for { + select { + case resp := <-recv.ch: + if resp.DERPMap != nil { + require.Len(t, frames, 1, "a tick of peer deltas must render as one frame") + assert.Len(t, frames[0].PeersChanged, len(storm)/2, "every added peer in the frame") + + return + } + + frames = append(frames, resp) + case <-deadline: + t.Fatal("never received the DERP map change") + } + } + }) + } +} + // TestBatcherPingSurvivesFullUpdate queues a ping and a full in one AddWork // call against real connections: the target renders the full and then a // ping-only frame, and no other node sees the ping. diff --git a/hscontrol/types/change/change.go b/hscontrol/types/change/change.go index 3a1a307dc..a0558cb9e 100644 --- a/hscontrol/types/change/change.go +++ b/hscontrol/types/change/change.go @@ -311,6 +311,67 @@ func DedupePolicyChanges(changes []Change) []Change { return out } +// isPeerDelta reports whether r only adds, refreshes or patches peers other +// than nodeID: [CoalescePeerDeltas] can fold such changes together. +func (r Change) isPeerDelta(nodeID types.NodeID) bool { + if r.IncludeSelf || r.IncludeDERPMap || r.IncludeDNS || r.IncludeDomain || + r.IncludePolicy || r.SendAllPeers || r.RequiresRuntimePeerComputation { + return false + } + + if r.OriginNode == nodeID || r.PingRequest != nil || + len(r.PeersRemoved) > 0 || len(r.DeletedNodes) > 0 { + return false + } + + return len(r.PeersChanged) > 0 || len(r.PeerPatches) > 0 +} + +// CoalescePeerDeltas folds each run of adjacent peer deltas in nodeID's +// changes for one tick into a single change, so a burst of joins renders as +// one [tailcfg.MapResponse] instead of one per change. Every change renders +// from live state after the drain, so a peer in the run's PeersChanged is sent +// whole and newer than any patch queued for it, and those patches are +// dropped. Runs never move across other changes, which keeps removals and +// policy recomputes in order. +func CoalescePeerDeltas(nodeID types.NodeID, changes []Change) []Change { + out := make([]Change, 0, len(changes)) + + for i := 0; i < len(changes); { + j := i + for j < len(changes) && changes[j].isPeerDelta(nodeID) { + j++ + } + + if j-i < 2 { + out = append(out, changes[i]) + i++ + + continue + } + + merged := Change{Reason: "coalesced peer changes"} + for _, c := range changes[i:j] { + merged.PeersChanged = append(merged.PeersChanged, c.PeersChanged...) + } + + merged.PeersChanged = uniqueNodeIDs(merged.PeersChanged) + + for _, c := range changes[i:j] { + for _, p := range c.PeerPatches { + if _, whole := slices.BinarySearch(merged.PeersChanged, types.NodeID(p.NodeID)); !whole { //nolint:gosec // NodeID types are equivalent + merged.PeerPatches = append(merged.PeerPatches, p) + } + } + } + + out = append(out, merged) + i = j + } + + 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 10a6de60c..739888616 100644 --- a/hscontrol/types/change/change_test.go +++ b/hscontrol/types/change/change_test.go @@ -486,6 +486,99 @@ func TestDedupePolicyChanges(t *testing.T) { } } +func TestCoalescePeerDeltas(t *testing.T) { + const self types.NodeID = 1 + + merged := func(peers []types.NodeID, patches ...*tailcfg.PeerChange) Change { + return Change{Reason: "coalesced peer changes", PeersChanged: peers, PeerPatches: patches} + } + patch := func(c Change) *tailcfg.PeerChange { return c.PeerPatches[0] } + + ping := PingNode(self, &tailcfg.PingRequest{URL: "https://example.com/ping"}) + + tests := []struct { + name string + changes []Change + want []Change + }{ + { + name: "single delta is unchanged", + changes: []Change{NodeAdded(2)}, + want: []Change{NodeAdded(2)}, + }, + { + name: "join storm folds into one change", + changes: []Change{NodeAdded(3), NodeOnline(3), NodeAdded(2), NodeOnline(2)}, + want: []Change{merged([]types.NodeID{2, 3})}, + }, + { + name: "patches for peers sent whole are dropped, others kept in order", + changes: []Change{NodeOnline(4), NodeAdded(2), NodeOffline(4), NodeOnline(2)}, + want: []Change{merged([]types.NodeID{2}, patch(NodeOnline(4)), patch(NodeOffline(4)))}, + }, + { + name: "runs do not cross other changes", + changes: []Change{ + NodeAdded(2), NodeOnline(2), PolicyChange(), NodeAdded(3), NodeRemoved(4), NodeOnline(5), ping, NodeOnline(6), + }, + want: []Change{ + merged([]types.NodeID{2}), PolicyChange(), NodeAdded(3), NodeRemoved(4), NodeOnline(5), ping, NodeOnline(6), + }, + }, + { + name: "the node's own change is a self update, not a peer delta", + changes: []Change{NodeAdded(2), NodeAdded(self), NodeAdded(3)}, + want: []Change{NodeAdded(2), NodeAdded(self), NodeAdded(3)}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, CoalescePeerDeltas(self, tt.changes)) + }) + } +} + +// TestCoalescePeerDeltasClassifiesEveryField classifies every [Change] field +// by whether a change setting it may fold into a run of peer deltas. A new +// field fails here until it is classified. +func TestCoalescePeerDeltasClassifiesEveryField(t *testing.T) { + const self = types.NodeID(1) + + folds := map[string]bool{ + "Reason": true, // logging only + "TargetNode": true, // a node's pending only holds its own + "OriginNode": false, // set to self: a self update + "IncludeSelf": false, + "IncludeDERPMap": false, + "IncludeDNS": false, + "IncludeDomain": false, + "IncludePolicy": false, + "PeersChanged": true, + "PeersRemoved": false, // must stay ordered against re-adds + "PeerPatches": true, + "SendAllPeers": false, + "DeletedNodes": false, + "RequiresRuntimePeerComputation": false, + "PingRequest": false, + } + + typ := reflect.TypeFor[Change]() + require.Len(t, folds, typ.NumField(), "classify every Change field") + + for field := range typ.Fields() { + t.Run(field.Name, func(t *testing.T) { + want, ok := folds[field.Name] + require.True(t, ok, "field %s is not classified", field.Name) + + c := Change{PeersChanged: []types.NodeID{2}} + setNonZero(t, reflect.ValueOf(&c).Elem().FieldByIndex(field.Index), self) + + assert.Equal(t, want, c.isPeerDelta(self)) + }) + } +} + func TestChange_Constructors(t *testing.T) { tests := []struct { name string