From f07a384fe5c67cfd842eefd9b6f0d9e7f704542c Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Tue, 27 Oct 2026 15:32:41 +0000 Subject: [PATCH] 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 | 15 ++++++ hscontrol/mapper/batcher_test.go | 78 ++++++++++++++++++++++++++++++++ hscontrol/mapper/node_conn.go | 17 ++++++- 3 files changed, 109 insertions(+), 1 deletion(-) diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 526bc2e03..53748729b 100644 --- a/hscontrol/mapper/batcher.go +++ b/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. diff --git a/hscontrol/mapper/batcher_test.go b/hscontrol/mapper/batcher_test.go index 5923cb5fe..7f87bed3f 100644 --- a/hscontrol/mapper/batcher_test.go +++ b/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") +} diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index 28291fedf..7009823a2 100644 --- a/hscontrol/mapper/node_conn.go +++ b/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 {