diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 9e6c2df8..ddeb8597 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -3,6 +3,7 @@ package mapper import ( "errors" "fmt" + "slices" "sync" "sync/atomic" "time" @@ -36,6 +37,14 @@ var mapResponseGenerated = promauto.NewCounterVec(prometheus.CounterOpts{ Help: "total count of mapresponses generated by response type", }, []string{"response_type"}) +// mapperChangesDropped tracks changes the batcher refused to fan out. +// Reason is bounded; "empty" covers IsEmpty changes dropped at ingress. +var mapperChangesDropped = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "headscale", + Name: "mapper_changes_dropped_total", + Help: "changes dropped by the batcher before recipient fan-out", +}, []string{"reason"}) + func NewBatcher(batchTime time.Duration, workers int, mapper *mapper) *Batcher { return &Batcher{ mapper: mapper, @@ -393,8 +402,21 @@ func (b *Batcher) RemoveNode(id types.NodeID, c chan<- *tailcfg.MapResponse) boo return false } -// AddWork queues a change to be processed by the batcher. +// AddWork queues a change to be processed by the batcher. Empty changes +// carry no work for any recipient and are dropped here so they never create +// pending entries or in-flight work. A PeersRemoved change is not empty, so +// deletion cleanup in addToBatch still runs. func (b *Batcher) AddWork(r ...change.Change) { + if slices.ContainsFunc(r, change.Change.IsEmpty) { + kept := slices.DeleteFunc(slices.Clone(r), change.Change.IsEmpty) + mapperChangesDropped.WithLabelValues("empty").Add(float64(len(r) - len(kept))) + r = kept + } + + if len(r) == 0 { + return + } + b.addToBatch(r...) } diff --git a/hscontrol/mapper/batcher_test.go b/hscontrol/mapper/batcher_test.go index 5350b064..c16e147c 100644 --- a/hscontrol/mapper/batcher_test.go +++ b/hscontrol/mapper/batcher_test.go @@ -2276,3 +2276,85 @@ func TestRemoveNodeChannelAlreadyRemoved(t *testing.T) { func unwrapBatcher(b *testBatcherWrapper) *Batcher { return b.Batcher } + +// TestAddWorkDropsEmptyChanges ensures an empty-only batch produces no +// pending entries on any recipient node. +func TestAddWorkDropsEmptyChanges(t *testing.T) { + lb := setupLightweightBatcher(t, 5, 16) + defer lb.cleanup() + + // A zero-value change.Change is empty: no peers changed, no patches, + // no full update, no removal. + empty := change.Change{} + require.True(t, empty.IsEmpty(), "zero change must be empty") + + lb.b.AddWork(empty) + + require.Equal(t, 0, countTotalPending(lb.b), + "empty change must not create pending entries on any node") + require.Equal(t, 0, countNodesPending(lb.b), + "empty change must not mark any node as having pending work") +} + +// TestAddWorkEmptyMixedWithReal ensures empty changes dropped from a batch +// do not affect delivery of the real changes alongside them. +func TestAddWorkEmptyMixedWithReal(t *testing.T) { + lb := setupLightweightBatcher(t, 3, 16) + defer lb.cleanup() + + added := change.NodeAdded(types.NodeID(42)) + empty := change.Change{} + + lb.b.AddWork(empty, added, empty) + + // The real change must land for every connected node; empties must not + // add anything. + require.Equal(t, 3, countNodesPending(lb.b), + "all 3 nodes should have the real change pending") + + for id := range lb.channels { + pending := getPendingForNode(lb.b, id) + require.Len(t, pending, 1, + "node %d must have exactly the one real change pending, no empties", id) + } +} + +// TestAddWorkFullUpdateUnaffectedByEmpty ensures that a full update in +// the batch still takes precedence and replaces pending state, regardless +// of any empty changes in the same batch. +func TestAddWorkFullUpdateUnaffectedByEmpty(t *testing.T) { + lb := setupLightweightBatcher(t, 3, 16) + defer lb.cleanup() + + lb.b.AddWork(change.Change{}, change.FullUpdate(), change.Change{}) + + for id := range lb.channels { + pending := getPendingForNode(lb.b, id) + require.Len(t, pending, 1, + "node %d pending should be the single full update", id) + require.True(t, pending[0].IsFull(), + "node %d pending entry must be the full update", id) + } +} + +// TestAddWorkPeersRemovedNotTreatedAsEmpty ensures the empty filter does +// not swallow a PeersRemoved change — deletion cleanup must still run, and +// surviving recipients must still see the removal. +func TestAddWorkPeersRemovedNotTreatedAsEmpty(t *testing.T) { + lb := setupLightweightBatcher(t, 4, 16) + defer lb.cleanup() + + // Remove node 4. + removal := change.NodeRemoved(types.NodeID(4)) + require.False(t, removal.IsEmpty(), "PeersRemoved change is not empty") + + lb.b.AddWork(removal) + + // Node 4 must be evicted from the batcher. + _, exists := lb.b.nodes.Load(types.NodeID(4)) + require.False(t, exists, "removed node must be evicted from batcher") + + // Surviving nodes must see the removal pending. + require.Equal(t, 3, countNodesPending(lb.b), + "surviving 3 nodes must have the removal pending") +} diff --git a/hscontrol/types/change/change.go b/hscontrol/types/change/change.go index ac5e550b..fc44c4c5 100644 --- a/hscontrol/types/change/change.go +++ b/hscontrol/types/change/change.go @@ -177,6 +177,11 @@ func (r Change) IsFull() bool { // This provides a bounded set of values suitable for Prometheus labels, // unlike [Change.Reason] which is free-form text for logging. func (r Change) Type() string { + // A suppressed update is a healthy outcome, not an unclassified one. + if r.IsEmpty() { + return "empty" + } + if r.IsFull() { return "full" } diff --git a/hscontrol/types/change/change_test.go b/hscontrol/types/change/change_test.go index 3f0d71a5..ea420ea7 100644 --- a/hscontrol/types/change/change_test.go +++ b/hscontrol/types/change/change_test.go @@ -561,9 +561,9 @@ func TestChange_Type(t *testing.T) { want: "ping", }, { - name: "empty is unknown", + name: "empty", response: Change{}, - want: "unknown", + want: "empty", }, }