mapper: keep one batched bundle per node in flight to preserve order

A saturated worker pool could deliver a later tick before an earlier one.

hscontrol/mapper/batch_coalesce_test.go

@@ -0,0 +1,47 @@
+package mapper
+
+import (
+	"testing"
+	"time"
+
+	"github.com/juanfont/headscale/hscontrol/types"
+	"github.com/juanfont/headscale/hscontrol/types/change"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestProcessBatchedChangesCoalescesWhenInFlight verifies that at most one
+// batched bundle per node is queued at a time. While a node's bundle is in
+// flight, a new tick's changes must stay in pending (coalesced) rather than
+// being queued as a second bundle that a non-FIFO worker could deliver out of
+// order.
+func TestProcessBatchedChangesCoalescesWhenInFlight(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)
+
+	// A bundle is in flight: the new change must be retained, not queued.
+	nc.inFlight.Store(true)
+	nc.appendPending(change.PolicyChange())
+	b.processBatchedChanges()
+
+	nc.pendingMu.Lock()
+	retained := len(nc.pending)
+	nc.pendingMu.Unlock()
+	assert.Equal(t, 1, retained, "pending must be retained while a bundle is in flight")
+
+	// No bundle in flight: the pending change is queued and the node marked
+	// in-flight.
+	nc.inFlight.Store(false)
+	b.processBatchedChanges()
+
+	nc.pendingMu.Lock()
+	drained := len(nc.pending)
+	nc.pendingMu.Unlock()
+	assert.Equal(t, 0, drained, "pending must be drained once queued")
+	assert.True(t, nc.inFlight.Load(), "queued bundle must mark the node in-flight")
+
+	require.NotNil(t, b)
+}

hscontrol/mapper/batcher.go

@@ -544,6 +544,9 @@ func (b *Batcher) worker(workerID int) {
 					}
 				}
 				nc.workMu.Unlock()
+
+				// Bundle delivered; allow the next tick's bundle to queue.
+				nc.inFlight.Store(false)
 			}
 		case <-b.done:
 			wlog.Debug().Msg("batcher shutting down, exiting worker")
@@ -645,6 +648,14 @@ func (b *Batcher) processBatchedChanges() {
 			return true
 		}
 
+		// Only one batched bundle per node may be in flight at a time. If the
+		// previous tick's bundle is still queued or processing, leave this
+		// tick's changes in pending; they are picked up once it completes. This
+		// keeps delivery ordered even when the worker pool is saturated.
+		if nc.inFlight.Load() {
+			return true
+		}
+
 		pending := nc.drainPending()
 		if len(pending) == 0 {
 			return true
@@ -654,8 +665,7 @@ func (b *Batcher) processBatchedChanges() {
 		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.
+		nc.inFlight.Store(true)
 		b.queueWork(work{changes: pending, nodeID: nodeID, resultCh: nil})
 
 		return true

hscontrol/mapper/node_conn.go

@@ -56,6 +56,13 @@ type multiChannelNodeConn struct {
 	// Range in [multiChannelNodeConn.computePeerDiff]).
 	workMu sync.Mutex
 
+	// inFlight is true while a batched work bundle for this node is queued or
+	// being processed. processBatchedChanges refuses to queue a second bundle
+	// while one is in flight (the new changes wait in pending), so a saturated
+	// worker pool cannot deliver tick N+1 before tick N: a non-FIFO workMu
+	// cannot reorder bundles that never coexist.
+	inFlight atomic.Bool
+
 	closeOnce   sync.Once
 	updateCount atomic.Int64