diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 052606a5..bcbcbae7 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -280,6 +280,11 @@ func (b *Batcher) AddNode( created: now, 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 newEntry.lastUsed.Store(now.Unix()) @@ -329,6 +334,11 @@ func (b *Batcher) AddNode( nodeConn.workMu.Lock() nodeConn.updateSentPeers(initialMap) 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 nlog.Error().Err(ErrInitialMapSendTimeout).Msg("initial map send timeout") 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 // on lastSentPeers (Clear+Store vs Range). 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() for _, ch := range w.changes { err := nc.change(ch) + if errors.Is(err, errNoReadyConnections) { + retry = append(retry, ch) + continue + } + if err != nil { b.workErrors.Add(1) wlog.Error().Err(err). @@ -545,6 +570,10 @@ func (b *Batcher) worker(workerID int) { } nc.workMu.Unlock() + if len(retry) > 0 { + nc.prependPending(retry...) + } + // Bundle delivered; allow the next tick's bundle to queue. nc.inFlight.Store(false) } diff --git a/hscontrol/mapper/initialmap_phantom_test.go b/hscontrol/mapper/initialmap_phantom_test.go index ba882894..362bd4f1 100644 --- a/hscontrol/mapper/initialmap_phantom_test.go +++ b/hscontrol/mapper/initialmap_phantom_test.go @@ -1,11 +1,76 @@ package mapper import ( + "errors" "testing" + "time" + "github.com/juanfont/headscale/hscontrol/types/change" "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 // 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 diff --git a/hscontrol/mapper/node_conn.go b/hscontrol/mapper/node_conn.go index fa0cd041..b91d2c7c 100644 --- a/hscontrol/mapper/node_conn.go +++ b/hscontrol/mapper/node_conn.go @@ -23,6 +23,16 @@ import ( // after this error because the data was never delivered to any client. 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. type connectionEntry struct { id string // unique connection ID @@ -32,6 +42,13 @@ type connectionEntry struct { stop func() lastUsed atomic.Int64 // Unix timestamp of last successful send 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. @@ -225,6 +242,17 @@ func (mc *multiChannelNodeConn) appendPending(changes ...change.Change) { 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. // Returns nil if there are no pending changes. func (mc *multiChannelNodeConn) drainPending() []change.Change { @@ -260,11 +288,27 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error { return errNoActiveConnections } - // Copy the slice so we can release the read lock before sending. - snapshot := make([]*connectionEntry, len(mc.connections)) - copy(snapshot, mc.connections) + // Copy only connections whose initial map has been delivered. + // A connection still awaiting its initial map receives one + // (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() + if len(snapshot) == 0 { + mc.log.Trace(). + Msg("send: connections present but none ready, requeue") + + return errNoReadyConnections + } + mc.log.Trace(). Int("total_connections", len(snapshot)). Msg("send: broadcasting")