mirror of
https://github.com/juanfont/headscale.git
synced 2026-10-28 01:06:50 +09:00
mapper: render each tick's peer deltas as one response
One frame per change let a join storm overflow the session channel in a single tick, and the batcher then dropped live clients as stale.
This commit is contained in:
1 parent
4933bb83d2
commit
7a2335dc2b
5 files changed
+221
-7
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user