mapper: bound how long one node can block a batcher worker

A node that drained slower than it was sent to held the worker for its
whole bundle; once its sends have waited 50ms, the rest waits a tick.

hscontrol/mapper/batcher.go

@@ -652,7 +652,22 @@ func (b *Batcher) worker(workerID int) {
 				var retry []change.Change
 
 				nc.workMu.Lock()
+				nc.sendBlocked.Store(0)
+
 				for i, ch := range w.changes {
+					// A node that drains slower than it is sent to must not
+					// hold the worker from the other nodes for its whole bundle.
+					if time.Duration(nc.sendBlocked.Load()) >= connectionSendWait {
+						retry = w.changes[i:]
+
+						wlog.Debug().
+							Uint64(zf.NodeID, w.nodeID.Uint64()).
+							Int("deferred_changes", len(retry)).
+							Msg("node drains slowly, deferring rest of bundle to next tick")
+
+						break
+					}
+
 					err := nc.change(ch)
 					if errors.Is(err, errNoReadyConnections) || errors.Is(err, errConnectionBackpressure) {
 						// Keep the unsent suffix ahead of newer pending changes.

hscontrol/mapper/batcher_test.go

@@ -3197,3 +3197,81 @@ func TestInvisiblePeerChangeSendsNoFrame(t *testing.T) {
 	deliver(other, change.PingNode(other.n.ID, &tailcfg.PingRequest{URL: "https://ping.test"}))
 	assert.NotNil(t, expectReceive(t, other.ch, "ping").PingRequest)
 }
+
+// TestWorkerDefersBundleOfSlowConnection pins that a connection which keeps
+// draining, but slower than a bundle is sent, does not hold the worker for
+// every send of the bundle. The node chooses how many responses its bundle
+// carries, while the worker holds workMu and is shared by every other node:
+// once the sends have waited connectionSendWait in total, the rest of the
+// bundle waits in pending for the next tick.
+func TestWorkerDefersBundleOfSlowConnection(t *testing.T) {
+	testData, cleanup := setupBatcherWithTestData(
+		t,
+		NewBatcherAndMapper,
+		1,
+		1,
+		normalBufferSize,
+	)
+	defer cleanup()
+
+	// A batcher that never ticks, so deferred changes stay in pending.
+	b := NewBatcher(time.Hour, 1, testData.Batcher.mapper)
+
+	b.Start()
+	defer b.Close()
+
+	id := testData.Nodes[0].n.ID
+	ch := make(chan *tailcfg.MapResponse) // unbuffered
+
+	nc := newMultiChannelNodeConn(id, b.mapper)
+	nc.addConnection(makeConnectionEntry("slow", ch))
+	b.nodes.Store(id, nc)
+
+	// Read one response per fifth of the wait: every send blocks, but well
+	// short of backpressure.
+	done := make(chan struct{})
+	defer close(done)
+
+	go func() {
+		for {
+			select {
+			case <-time.After(connectionSendWait / 5):
+			case <-done:
+				return
+			}
+
+			select {
+			case <-ch:
+			case <-done:
+				return
+			}
+		}
+	}()
+
+	bundle := make([]change.Change, 100)
+	for i := range bundle {
+		bundle[i] = change.DERPMap()
+		bundle[i].Reason = fmt.Sprintf("derp map update %d", i)
+	}
+
+	start := time.Now()
+
+	nc.inFlight.Store(true)
+	b.queueWork(work{changes: bundle, nodeID: id})
+
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.False(c, nc.inFlight.Load())
+	}, 5*time.Second, time.Millisecond, "worker should finish with the bundle")
+
+	assert.Less(t, time.Since(start), 10*connectionSendWait,
+		"the sends of one bundle should not each wait for the connection")
+	assert.Equal(t, 1, nc.getActiveConnectionCount(),
+		"a connection that keeps draining is not stale")
+
+	pending := getPendingForNode(b, id)
+	sent := len(bundle) - len(pending)
+	assert.Positive(t, sent, "sends should go through until the wait is used up")
+	assert.NotEmpty(t, pending, "the rest of the bundle should be deferred")
+	assert.Equal(t, bundle[sent:], pending,
+		"what was not sent should be pending, in order")
+}

hscontrol/mapper/node_conn.go

@@ -48,6 +48,11 @@ var errConnectionBackpressure = errors.New("connection queue full")
 // Match the initial-map timeout while staying below reconnect grace and client keepalive.
 const connectionStallTimeout = 5 * time.Second
 
+// connectionSendWait is how long one send waits for room on a connection
+// before it counts as backpressure, and how long all sends of one bundle
+// may wait before [Batcher.worker] retries the rest next tick.
+const connectionSendWait = 50 * time.Millisecond
+
 // A full snapshot supersedes state backlog past this size; pings remain
 // one-shot work.
 const maxPendingSize = 256
@@ -149,6 +154,13 @@ type multiChannelNodeConn struct {
 	// cannot reorder bundles that never coexist.
 	inFlight atomic.Bool
 
+	// sendBlocked is how long [multiChannelNodeConn.send] has waited on
+	// connections during the current bundle. The node decides how fast its
+	// connections drain and, through its own map requests, how many
+	// responses a bundle carries, while the worker holding workMu serves
+	// every node.
+	sendBlocked atomic.Int64
+
 	closeOnce   sync.Once
 	updateCount atomic.Int64
 
@@ -486,6 +498,9 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error {
 		newestAccepted bool
 	)
 
+	start := time.Now()
+	defer func() { mc.sendBlocked.Add(int64(time.Since(start))) }()
+
 	for _, conn := range snapshot {
 		err := conn.send(conn.withSSHPolicyDelta(conn.withSelfDelta(data)))
 		if errors.Is(err, errConnectionBackpressure) {
@@ -588,7 +603,7 @@ func (entry *connectionEntry) send(data *tailcfg.MapResponse) error {
 	//
 	// Use a timer rather than time.After so the timeout is explicitly released
 	// on the fast path; both are GC-recoverable since Go 1.23.
-	timer := time.NewTimer(50 * time.Millisecond) //nolint:mnd
+	timer := time.NewTimer(connectionSendWait)
 	defer timer.Stop()
 
 	select {