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

@@ -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

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")