diff --git a/hscontrol/mapper/batch_coalesce_test.go b/hscontrol/mapper/batch_coalesce_test.go index d89a48766..bd7ae483d 100644 --- a/hscontrol/mapper/batch_coalesce_test.go +++ b/hscontrol/mapper/batch_coalesce_test.go @@ -8,6 +8,7 @@ import ( "github.com/juanfont/headscale/hscontrol/types/change" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "tailscale.com/tailcfg" ) // TestProcessBatchedChangesCoalescesWhenInFlight verifies that at most one @@ -20,6 +21,7 @@ func TestProcessBatchedChangesCoalescesWhenInFlight(t *testing.T) { id := types.NodeID(1) nc := newMultiChannelNodeConn(id, nil) + nc.addConnection(makeConnectionEntry("conn-1", make(chan *tailcfg.MapResponse, 1))) b.nodes.Store(id, nc) // A bundle is in flight: the new change must be retained, not queued. @@ -45,3 +47,25 @@ func TestProcessBatchedChangesCoalescesWhenInFlight(t *testing.T) { require.NotNil(t, b) } + +// TestProcessBatchedChangesSkipsNodesWithoutConnection drains a node that +// kept its batcher entry after its last stream closed. Rendering its changes +// is wasted, since send drops the result and its next stream opens with a +// full map, so no work may be queued for it. +func TestProcessBatchedChangesSkipsNodesWithoutConnection(t *testing.T) { + b := NewBatcher(50*time.Millisecond, 2, nil) // not started: no ticker, no workers + + id := types.NodeID(1) + nc := newMultiChannelNodeConn(id, nil) + b.nodes.Store(id, nc) + + nc.appendPending(change.PolicyChange()) + b.processBatchedChanges() + + nc.pendingMu.Lock() + pending := len(nc.pending) + nc.pendingMu.Unlock() + assert.Equal(t, 0, pending, "a node without a connection must not accumulate pending changes") + assert.Empty(t, b.workCh, "no work may be queued for a node without a connection") + assert.False(t, nc.inFlight.Load()) +} diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index a1e8d1fec..ad63c42c1 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -745,7 +745,7 @@ func (b *Batcher) processBatchedChanges() { return true } - pending := nc.drainPending() + pending := nc.takePending() if len(pending) == 0 { return true } diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index 1203cd316..cba71ce5f 100644 --- a/hscontrol/mapper/node_conn.go +++ b/hscontrol/mapper/node_conn.go @@ -319,6 +319,25 @@ func (mc *multiChannelNodeConn) drainPending() []change.Change { return p } +// takePending drains the pending changes for rendering. A node without a +// connection gets none and its changes are dropped: send would discard what +// they render, and its next connection opens with a full map rendered after +// it registered. The read lock keeps that registration out of the gap between +// the check and the drain. +func (mc *multiChannelNodeConn) takePending() []change.Change { + mc.mutex.RLock() + defer mc.mutex.RUnlock() + + p := mc.drainPending() + if len(mc.connections) == 0 { + mapperChangesDropped.WithLabelValues("no_connection").Add(float64(len(p))) + + return nil + } + + return p +} + // send broadcasts data to all active connections for the node. // // To avoid holding the write lock during potentially slow sends (each stale