From ee3a5600c7b90fa9c6e3fac6fcb0a0ce31f5c42c Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Tue, 27 Oct 2026 15:26:26 +0000 Subject: [PATCH] mapper: send nothing while any stream awaits its initial map Sent to the ready streams alone, a change counted as delivered and never reached the new stream; the worker retries it once the map is out. --- hscontrol/mapper/initialmap_phantom_test.go | 236 ++++++++++++++++++++ hscontrol/mapper/node_conn.go | 54 +++-- 2 files changed, 261 insertions(+), 29 deletions(-) diff --git a/hscontrol/mapper/initialmap_phantom_test.go b/hscontrol/mapper/initialmap_phantom_test.go index 7d0158afc..ec81b79da 100644 --- a/hscontrol/mapper/initialmap_phantom_test.go +++ b/hscontrol/mapper/initialmap_phantom_test.go @@ -2,11 +2,16 @@ package mapper import ( "errors" + "fmt" "reflect" + "slices" + "sync" "testing" "time" + "github.com/juanfont/headscale/hscontrol/types" "github.com/juanfont/headscale/hscontrol/types/change" + "github.com/stretchr/testify/assert" "tailscale.com/tailcfg" ) @@ -73,6 +78,237 @@ func TestUnreadyConnectionDefersBroadcastsUntilInitialMap(t *testing.T) { } } +// TestUnreadyConnectionBesideReadyOneHoldsBroadcasts pins that a node's change +// reaches none of its streams while one of them still awaits its initial map, +// which may predate the change. Sent to the ready stream alone, the change +// would count as delivered and never reach the new one. +func TestUnreadyConnectionBesideReadyOneHoldsBroadcasts(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] + + nc := newMultiChannelNodeConn(targetNode.n.ID, batcher.mapper) + nc.addConnection(makeConnectionEntry("test-ready", targetNode.ch)) + + pendingCh := make(chan *tailcfg.MapResponse, normalBufferSize) + pending := makeConnectionEntry("test-unready", pendingCh) + pending.pendingInitial.Store(true) + nc.addConnection(pending) + batcher.nodes.Store(targetNode.n.ID, nc) + + retryChange := change.NodeAdded(peerNode.n.ID) + + err := nc.change(retryChange) + if !errors.Is(err, errNoReadyConnections) { + t.Fatalf( + "change with a connection awaiting its initial map: want errNoReadyConnections, got %v", + err, + ) + } + + for name, ch := range map[string]chan *tailcfg.MapResponse{"ready": targetNode.ch, "unready": pendingCh} { + select { + case resp := <-ch: + t.Fatalf( + "%s connection received a frame while a stream awaits its initial map: %+v", + name, + resp, + ) + default: + } + } + + pending.pendingInitial.Store(false) + + err = nc.change(retryChange) + if err != nil { + t.Fatalf("change on ready connections: %v", err) + } + + for name, ch := range map[string]chan *tailcfg.MapResponse{"ready": targetNode.ch, "new": pendingCh} { + select { + case resp := <-ch: + if len(resp.PeersChanged) == 0 { + t.Fatalf("%s connection: expected PeersChanged delta, got %+v", + name, resp) + } + default: + t.Fatalf("%s connection did not receive the retried change", name) + } + } +} + +// onlineRecorder keeps the peer online states a stream receives, in order. +type onlineRecorder struct { + mu sync.Mutex + got []string +} + +func (r *onlineRecorder) record(resp *tailcfg.MapResponse) { + r.mu.Lock() + defer r.mu.Unlock() + + for _, p := range resp.PeersChangedPatch { + if p.Online != nil { + r.got = append(r.got, fmt.Sprintf("%d:%t", p.NodeID, *p.Online)) + } + } +} + +func (r *onlineRecorder) states() []string { + r.mu.Lock() + defer r.mu.Unlock() + + return slices.Clone(r.got) +} + +// TestHeldChangesKeepOrderWhenStreamTurnsReady reconnects a node mid-bundle: +// its new stream turns ready while the bundle's first change is being sent to +// the node's older streams. Every stream must get every change once, in order, +// and end on the newest state. A change of another kind sits between the two +// peer changes, so the bundle keeps them as separate changes. +func TestHeldChangesKeepOrderWhenStreamTurnsReady(t *testing.T) { + tests := []struct { + name string + bundle func(p, q types.NodeID) []change.Change + want func(p, q types.NodeID) []string + }{ + { + name: "online_then_offline", + bundle: func(p, _ types.NodeID) []change.Change { + return []change.Change{ + change.NodeOnline(p), change.DERPMap(), change.NodeOffline(p), + } + }, + want: func(p, _ types.NodeID) []string { + return []string{fmt.Sprintf("%d:true", p), fmt.Sprintf("%d:false", p)} + }, + }, + { + name: "two_peers_online", + bundle: func(p, q types.NodeID) []change.Change { + return []change.Change{ + change.NodeOnline(p), change.DERPMap(), change.NodeOnline(q), + } + }, + want: func(p, q types.NodeID) []string { + return []string{fmt.Sprintf("%d:true", p), fmt.Sprintf("%d:true", q)} + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + testData, cleanup := setupBatcherWithTestData( + t, + NewBatcherAndMapper, + 1, + 3, + normalBufferSize, + ) + defer cleanup() + + batcher := testData.Batcher.Batcher + p, q, target := testData.Nodes[0].n.ID, testData.Nodes[1].n.ID, testData.Nodes[2].n.ID + + // Two older streams, unbuffered so a send blocks until it is read, + // and the new one, still awaiting its initial map. + firstCh := make(chan *tailcfg.MapResponse) + secondCh := make(chan *tailcfg.MapResponse) + newCh := make(chan *tailcfg.MapResponse, normalBufferSize) + fresh := makeConnectionEntry("new", newCh) + fresh.pendingInitial.Store(true) + + nc := newMultiChannelNodeConn(target, batcher.mapper) + nc.addConnection(makeConnectionEntry("first", firstCh)) + nc.addConnection(makeConnectionEntry("second", secondCh)) + nc.addConnection(fresh) + batcher.nodes.Store(target, nc) + + // The new stream's initial map goes out once the first stream + // takes a frame, so while the second one waits for its own; or a + // little later if no frame is sent first. + var once sync.Once + + ready := make(chan struct{}) + markReady := func() { + once.Do(func() { + fresh.pendingInitial.Store(false) + close(ready) + }) + } + + timer := time.AfterFunc(200*time.Millisecond, markReady) + defer timer.Stop() + + var firstGot, secondGot, newGot onlineRecorder + + stop := make(chan struct{}) + defer close(stop) + + read := func(ch <-chan *tailcfg.MapResponse, r *onlineRecorder, onFrame func()) { + for { + select { + case resp := <-ch: + onFrame() + r.record(resp) + case <-stop: + return + } + } + } + + go read(firstCh, &firstGot, markReady) + go read(newCh, &newGot, func() {}) + go func() { + select { + case <-ready: + read(secondCh, &secondGot, func() {}) + case <-stop: + } + }() + + batcher.AddWork(tt.bundle(p, q)...) + + want := tt.want(p, q) + streams := map[string]*onlineRecorder{ + "first": &firstGot, + "second": &secondGot, + "new": &newGot, + } + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + for name, r := range streams { + assert.Len(c, r.states(), len(want), name) + } + }, 5*time.Second, 10*time.Millisecond) + + assert.Never(t, func() bool { + for _, r := range streams { + if len(r.states()) > len(want) { + return true + } + } + + return false + }, 100*time.Millisecond, 10*time.Millisecond, "no duplicate or late frame follows") + + for name, r := range streams { + assert.Equal(t, want, r.states(), "%s stream", name) + } + }) + } +} + // 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 5af1399c0..28291fedf 100644 --- a/hscontrol/mapper/node_conn.go +++ b/hscontrol/mapper/node_conn.go @@ -31,14 +31,15 @@ const maxConcurrentMapSessionsPerNode = 4 // 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. +// errNoReadyConnections is returned by [multiChannelNodeConn.send], having +// sent nothing, when any of the node's connections is still waiting for its +// initial map ([Batcher.AddNode] has registered it but not yet delivered the +// first full response). Sending a delta to it 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, and on every connection: 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") // errConnectionBackpressure keeps unsent changes queued without declaring a busy client dead. @@ -67,7 +68,7 @@ type connectionEntry struct { // 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 + // is delivered. Broadcast sends to the node wait for it, 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 @@ -435,9 +436,9 @@ func (mc *multiChannelNodeConn) takePending() []change.Change { // To avoid holding the write lock during bounded channel waits, the method // snapshots connections under a read lock, sends without any lock held, then // write-locks only to remove failures. New connections added between the -// snapshot and cleanup are safe: -// they receive a full initial map via [Batcher.AddNode], so missing this update -// causes no data loss. +// snapshot and cleanup are safe: [Batcher.AddNode] generates their initial +// map under workMu, after this bundle, so missing this update causes no data +// loss. func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error { if data == nil { return nil @@ -454,28 +455,23 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error { return errNoActiveConnections } - // 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 { + // A connection still awaiting its initial map must not get this + // update first, and the map may predate it: send to none until the + // map is out, see [errNoReadyConnections]. + if slices.ContainsFunc(mc.connections, func(conn *connectionEntry) bool { + return conn.pendingInitial.Load() + }) { + mc.mutex.RUnlock() mc.log.Trace(). - Msg("send: connections present but none ready, requeue") + Msg("send: a connection awaits its initial map, requeue") return errNoReadyConnections } + snapshot := slices.Clone(mc.connections) + + mc.mutex.RUnlock() + mc.log.Trace(). Int("total_connections", len(snapshot)). Msg("send: broadcasting")