mirror of
https://github.com/juanfont/headscale.git
synced 2026-10-08 15:50:06 +09:00
mapper: gate broadcast sends until a connection's initial map is delivered
AddNode registers the connection before sending the initial map, so a batched delta could land first and become the stream's first frame; Tailscale clients then kill the poll with "initial MapResponse lacked Node" and a retry loop under steady change traffic leaves the netmap empty. Skip connections awaiting their initial map and retry the change next tick — the in-flight map may predate it, so it cannot be dropped. Retries are prepended to keep patch order.
This commit is contained in:
committed by
Kristoffer Dalby
parent
7918187e7a
commit
f4eeb94b1c
@@ -280,6 +280,11 @@ func (b *Batcher) AddNode(
|
|||||||
created: now,
|
created: now,
|
||||||
stop: stop,
|
stop: stop,
|
||||||
}
|
}
|
||||||
|
// Block broadcast sends to this connection until its initial map
|
||||||
|
// is delivered below, so a delta cannot become the stream's first
|
||||||
|
// frame — clients reject streams whose first frame lacks the self
|
||||||
|
// node.
|
||||||
|
newEntry.pendingInitial.Store(true)
|
||||||
// Initialize last used timestamp
|
// Initialize last used timestamp
|
||||||
newEntry.lastUsed.Store(now.Unix())
|
newEntry.lastUsed.Store(now.Unix())
|
||||||
|
|
||||||
@@ -329,6 +334,11 @@ func (b *Batcher) AddNode(
|
|||||||
nodeConn.workMu.Lock()
|
nodeConn.workMu.Lock()
|
||||||
nodeConn.updateSentPeers(initialMap)
|
nodeConn.updateSentPeers(initialMap)
|
||||||
nodeConn.workMu.Unlock()
|
nodeConn.workMu.Unlock()
|
||||||
|
|
||||||
|
// Open the connection for broadcast sends now that the initial
|
||||||
|
// map is the stream's first frame; send() requeued any changes
|
||||||
|
// that arrived in the meantime.
|
||||||
|
newEntry.pendingInitial.Store(false)
|
||||||
case <-time.After(5 * time.Second): //nolint:mnd
|
case <-time.After(5 * time.Second): //nolint:mnd
|
||||||
nlog.Error().Err(ErrInitialMapSendTimeout).Msg("initial map send timeout")
|
nlog.Error().Err(ErrInitialMapSendTimeout).Msg("initial map send timeout")
|
||||||
nlog.Debug().Caller().Dur("timeout.duration", 5*time.Second). //nolint:mnd
|
nlog.Debug().Caller().Dur("timeout.duration", 5*time.Second). //nolint:mnd
|
||||||
@@ -532,9 +542,24 @@ func (b *Batcher) worker(workerID int) {
|
|||||||
// finish — preventing out-of-order delivery and races
|
// finish — preventing out-of-order delivery and races
|
||||||
// on lastSentPeers (Clear+Store vs Range).
|
// on lastSentPeers (Clear+Store vs Range).
|
||||||
if nc, exists := b.nodes.Load(w.nodeID); exists && nc != nil {
|
if nc, exists := b.nodes.Load(w.nodeID); exists && nc != nil {
|
||||||
|
// Changes that found only connections still awaiting
|
||||||
|
// their initial map are retried next tick: the
|
||||||
|
// in-flight initial map may have been generated from
|
||||||
|
// a snapshot older than the change, so dropping them
|
||||||
|
// would lose updates. Collected and prepended as a
|
||||||
|
// group to keep their order ahead of newer pending
|
||||||
|
// changes — order matters for stateful patches like
|
||||||
|
// online/offline.
|
||||||
|
var retry []change.Change
|
||||||
|
|
||||||
nc.workMu.Lock()
|
nc.workMu.Lock()
|
||||||
for _, ch := range w.changes {
|
for _, ch := range w.changes {
|
||||||
err := nc.change(ch)
|
err := nc.change(ch)
|
||||||
|
if errors.Is(err, errNoReadyConnections) {
|
||||||
|
retry = append(retry, ch)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
b.workErrors.Add(1)
|
b.workErrors.Add(1)
|
||||||
wlog.Error().Err(err).
|
wlog.Error().Err(err).
|
||||||
@@ -545,6 +570,10 @@ func (b *Batcher) worker(workerID int) {
|
|||||||
}
|
}
|
||||||
nc.workMu.Unlock()
|
nc.workMu.Unlock()
|
||||||
|
|
||||||
|
if len(retry) > 0 {
|
||||||
|
nc.prependPending(retry...)
|
||||||
|
}
|
||||||
|
|
||||||
// Bundle delivered; allow the next tick's bundle to queue.
|
// Bundle delivered; allow the next tick's bundle to queue.
|
||||||
nc.inFlight.Store(false)
|
nc.inFlight.Store(false)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,11 +1,76 @@
|
|||||||
package mapper
|
package mapper
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/juanfont/headscale/hscontrol/types/change"
|
||||||
"tailscale.com/tailcfg"
|
"tailscale.com/tailcfg"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// TestUnreadyConnectionDefersBroadcastsUntilInitialMap pins the ordering fix
|
||||||
|
// for the "initial MapResponse lacked Node" client failure: a connection
|
||||||
|
// registered by [Batcher.AddNode] but still waiting for its initial map must
|
||||||
|
// not receive broadcast deltas — a delta as the stream's first frame makes the
|
||||||
|
// Tailscale client tear down the poll. The change is requeued, not dropped,
|
||||||
|
// because the in-flight initial map may have been generated from a snapshot
|
||||||
|
// older than the change.
|
||||||
|
func TestUnreadyConnectionDefersBroadcastsUntilInitialMap(t *testing.T) {
|
||||||
|
testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, 2, normalBufferSize)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
batcher := testData.Batcher.Batcher
|
||||||
|
peerNode := &testData.Nodes[0]
|
||||||
|
targetNode := &testData.Nodes[1]
|
||||||
|
|
||||||
|
// Register the target's connection by hand in the state AddNode leaves it
|
||||||
|
// in between registering the channel and delivering the initial map.
|
||||||
|
nc := newMultiChannelNodeConn(targetNode.n.ID, batcher.mapper)
|
||||||
|
entry := &connectionEntry{
|
||||||
|
id: "test-unready",
|
||||||
|
c: targetNode.ch,
|
||||||
|
version: tailcfg.CapabilityVersion(100),
|
||||||
|
created: time.Now(),
|
||||||
|
}
|
||||||
|
entry.pendingInitial.Store(true)
|
||||||
|
nc.addConnection(entry)
|
||||||
|
batcher.nodes.Store(targetNode.n.ID, nc)
|
||||||
|
|
||||||
|
// A broadcast lands while the initial map is still in flight: nothing may
|
||||||
|
// reach the channel, and the caller must be told to retry
|
||||||
|
// (the worker prepends such changes back onto pending).
|
||||||
|
retryChange := change.NodeAdded(peerNode.n.ID)
|
||||||
|
|
||||||
|
err := nc.change(retryChange)
|
||||||
|
if !errors.Is(err, errNoReadyConnections) {
|
||||||
|
t.Fatalf("change on unready connection: want errNoReadyConnections, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case resp := <-targetNode.ch:
|
||||||
|
t.Fatalf("unready connection received a frame before its initial map: %+v", resp)
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
// Once the initial map is delivered, the retried change goes out.
|
||||||
|
entry.pendingInitial.Store(false)
|
||||||
|
|
||||||
|
err = nc.change(retryChange)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("change on ready connection: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case resp := <-targetNode.ch:
|
||||||
|
if len(resp.PeersChanged) == 0 {
|
||||||
|
t.Fatalf("expected PeersChanged delta after readiness, got %+v", resp)
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
t.Fatal("ready connection did not receive the retried change")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestSyncInitialMapNoPhantomPeersOnTimeout ensures the synchronous initial-map
|
// TestSyncInitialMapNoPhantomPeersOnTimeout ensures the synchronous initial-map
|
||||||
// path does not record peers as sent until the map is actually delivered. When
|
// path does not record peers as sent until the map is actually delivered. When
|
||||||
// the AddNode channel send times out, the client received nothing, so
|
// the AddNode channel send times out, the client received nothing, so
|
||||||
|
|||||||
@@ -23,6 +23,16 @@ import (
|
|||||||
// after this error because the data was never delivered to any client.
|
// after this error because the data was never delivered to any client.
|
||||||
var errNoActiveConnections = errors.New("no active connections")
|
var errNoActiveConnections = errors.New("no active connections")
|
||||||
|
|
||||||
|
// errNoReadyConnections is returned by [multiChannelNodeConn.send] when the
|
||||||
|
// node's only connections are still waiting for their initial map
|
||||||
|
// ([Batcher.AddNode] has registered them but not yet delivered the first full
|
||||||
|
// response). Sending a delta now would make it the stream's first frame, which
|
||||||
|
// Tailscale clients reject ("initial MapResponse lacked Node") — tearing down
|
||||||
|
// the poll. Unlike [errNoActiveConnections], the change must be retried: the
|
||||||
|
// in-flight initial map may have been generated from a snapshot older than
|
||||||
|
// the change, so dropping it would lose the update.
|
||||||
|
var errNoReadyConnections = errors.New("no connections ready for updates")
|
||||||
|
|
||||||
// connectionEntry represents a single connection to a node.
|
// connectionEntry represents a single connection to a node.
|
||||||
type connectionEntry struct {
|
type connectionEntry struct {
|
||||||
id string // unique connection ID
|
id string // unique connection ID
|
||||||
@@ -32,6 +42,13 @@ type connectionEntry struct {
|
|||||||
stop func()
|
stop func()
|
||||||
lastUsed atomic.Int64 // Unix timestamp of last successful send
|
lastUsed atomic.Int64 // Unix timestamp of last successful send
|
||||||
closed atomic.Bool // Indicates if this connection has been closed
|
closed atomic.Bool // Indicates if this connection has been closed
|
||||||
|
|
||||||
|
// pendingInitial is set by [Batcher.AddNode] while this
|
||||||
|
// connection's initial map is still in flight, and cleared once it
|
||||||
|
// is delivered. Broadcast sends skip such connections so a delta
|
||||||
|
// can never become the stream's first frame ahead of the initial
|
||||||
|
// map. The zero value means the connection is ready.
|
||||||
|
pendingInitial atomic.Bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// multiChannelNodeConn manages multiple concurrent connections for a single node.
|
// multiChannelNodeConn manages multiple concurrent connections for a single node.
|
||||||
@@ -225,6 +242,17 @@ func (mc *multiChannelNodeConn) appendPending(changes ...change.Change) {
|
|||||||
mc.pendingMu.Unlock()
|
mc.pendingMu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// prependPending puts changes at the head of the pending list, ahead of
|
||||||
|
// anything queued since. Used to retry changes that could not be
|
||||||
|
// delivered yet (initial map in flight): they were emitted before the
|
||||||
|
// currently pending ones, and order matters for stateful patches like
|
||||||
|
// online/offline.
|
||||||
|
func (mc *multiChannelNodeConn) prependPending(changes ...change.Change) {
|
||||||
|
mc.pendingMu.Lock()
|
||||||
|
mc.pending = append(changes, mc.pending...)
|
||||||
|
mc.pendingMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
// drainPending atomically removes and returns all pending changes.
|
// drainPending atomically removes and returns all pending changes.
|
||||||
// Returns nil if there are no pending changes.
|
// Returns nil if there are no pending changes.
|
||||||
func (mc *multiChannelNodeConn) drainPending() []change.Change {
|
func (mc *multiChannelNodeConn) drainPending() []change.Change {
|
||||||
@@ -260,11 +288,27 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error {
|
|||||||
return errNoActiveConnections
|
return errNoActiveConnections
|
||||||
}
|
}
|
||||||
|
|
||||||
// Copy the slice so we can release the read lock before sending.
|
// Copy only connections whose initial map has been delivered.
|
||||||
snapshot := make([]*connectionEntry, len(mc.connections))
|
// A connection still awaiting its initial map receives one
|
||||||
copy(snapshot, mc.connections)
|
// (generated from the current snapshot) from [Batcher.AddNode];
|
||||||
|
// pushing this update at it now would deliver a delta as the
|
||||||
|
// stream's first frame.
|
||||||
|
snapshot := make([]*connectionEntry, 0, len(mc.connections))
|
||||||
|
|
||||||
|
for _, conn := range mc.connections {
|
||||||
|
if !conn.pendingInitial.Load() {
|
||||||
|
snapshot = append(snapshot, conn)
|
||||||
|
}
|
||||||
|
}
|
||||||
mc.mutex.RUnlock()
|
mc.mutex.RUnlock()
|
||||||
|
|
||||||
|
if len(snapshot) == 0 {
|
||||||
|
mc.log.Trace().
|
||||||
|
Msg("send: connections present but none ready, requeue")
|
||||||
|
|
||||||
|
return errNoReadyConnections
|
||||||
|
}
|
||||||
|
|
||||||
mc.log.Trace().
|
mc.log.Trace().
|
||||||
Int("total_connections", len(snapshot)).
|
Int("total_connections", len(snapshot)).
|
||||||
Msg("send: broadcasting")
|
Msg("send: broadcasting")
|
||||||
|
|||||||
Reference in New Issue
Block a user