servertest: exercise control plane at explicit scales

Cover policy, updates and reconnects with convergence and resource reports.

hscontrol/servertest/README.md

@@ -0,0 +1,109 @@
+# Server stress tests
+
+`TestScale` exercises a real Headscale server, SQLite, policy compilation,
+the NodeStore, mapper workers, Noise, and Tailscale control clients over the
+existing in-memory network. Every registered node has a live control client.
+The test cases and their policies are together in `scale_test.go`.
+
+The ordinary test suite runs all five policies with 12 nodes. Larger runs
+require `HEADSCALE_STRESS_NODES`; setting it replaces the small node count.
+Nothing large is enabled by `go test ./...` or by omitting `-short`.
+
+| Case               | Expected visibility                                          | Main pressure                                      |
+| ------------------ | ------------------------------------------------------------ | -------------------------------------------------- |
+| `allow_all`        | Everyone sees everyone                                       | Dense graph, map size, broadcast fan-out           |
+| `isolated`         | No peers                                                     | Registration and state updates without peer output |
+| `personal_devices` | Up to ten devices belonging to the same user                 | Per-node policy compilation and isolation          |
+| `shared_service`   | One tagged service sees everyone; everyone else sees only it | Uneven fan-out and tag identity                    |
+| `personal_ssh`     | Personal devices, with SSH to the same user's devices        | Per-node SSH policy generation                     |
+
+Each case enrolls its clients, sends three Hostinfo bursts, revokes access,
+restores access, and reconnects one client per user. Bursts change one node
+per ten-node user; the final user can have fewer nodes. Registration uses
+batches of at most 32 concurrent logins. Mapper workers are
+`min(4, GOMAXPROCS)`, with the harness's 50 ms mapper delay, 30-frame session
+buffer. Explicit scale runs use the server's default NodeStore batching;
+the routine 12-node smoke run shortens its batch wait to 10 ms.
+
+Completion means the intended state reached every affected client, with the
+exact permitted peer IDs, fresh OS markers, and online peers. Bounded
+IPv4/IPv6 filter probes check allowed TCP traffic and denied ports, protocols,
+and cross-user access. SSH checks the accept action, root mapping, and exact
+same-user principal addresses. After reconnecting, a new policy generation
+broadcast checks delivery on every recovered stream, including isolated nodes.
+Policy changes carry a new node attribute so an old map cannot satisfy a
+policy wait. Revocation also checks that packet filter rules are empty.
+Each phase checks both client streams and server connections and rejects
+mapper work errors. Convergence has a shared two-minute deadline per wait,
+and existing login helpers retain their ten-second registration timeout.
+There are no sleep-based completion checks or automatic reconnect retries.
+
+## Run
+
+Use the pinned Go toolchain from `nix develop`, as for the other server tests.
+
+```sh
+# Fast coverage of every case; this also runs in the ordinary suite.
+go test ./hscontrol/servertest -run '^TestScale$' -count=1 -v
+
+# Select a scale and one policy. All nodes are real connected control clients.
+HEADSCALE_STRESS_NODES=500 GOMAXPROCS=4 \
+  go test ./hscontrol/servertest -run '^TestScale/500_nodes/personal_devices$' \
+  -count=1 -v -timeout=20m
+
+HEADSCALE_STRESS_NODES=1000 GOMAXPROCS=4 \
+  go test ./hscontrol/servertest -run '^TestScale/1000_nodes/shared_service$' \
+  -count=1 -v -timeout=30m
+
+HEADSCALE_STRESS_NODES=10000 GOMAXPROCS=4 \
+  go test ./hscontrol/servertest -run '^TestScale/10000_nodes/personal_devices$' \
+  -count=1 -v -timeout=60m
+
+# A compute-constrained run: one Go execution thread and one mapper worker.
+HEADSCALE_STRESS_NODES=500 GOMAXPROCS=1 GOMEMLIMIT=512MiB \
+  go test ./hscontrol/servertest -run '^TestScale/500_nodes/personal_devices$' \
+  -count=1 -v -timeout=30m
+```
+
+Omit the final policy component of `-run` to run every policy at that size.
+Choose large dense runs deliberately: 10,000 all-to-all clients hold
+99,990,000 directed peer entries before accounting for serialized maps and
+other state. A sparse output graph still requires policy/peer calculations;
+it does not guarantee linear CPU use. `GOMEMLIMIT` is a soft Go heap target,
+not an RSS cap. Go's resource settings constrain server and clients together.
+
+## Read results and find congestion
+
+Each phase logs wall time, allocated bytes during the phase, ending heap
+bytes, GC cycles, goroutines, registered nodes, live clients, and visible
+peer entries. Ending heap is not peak memory or RSS. These numbers cover
+the whole test process, including clients and assertions. They are useful
+for comparing identical workloads and finding limits, not server-only
+latency or memory claims. Old client netmap histories are disabled; the
+current maps remain, as they do in connected clients.
+
+Repeat one selected workload with fixed toolchain, node count, runtime
+settings, and hardware. Disable Go's test cache with `-count=1`, run each
+sample in a fresh process, keep unrelated heavy jobs off the host,
+and save the commit ID, `go version`, CPU model, and command with the log.
+Keep enrollment and steady-state phase results separate. Failures and
+timeouts are results, not samples to silently retry or discard.
+
+```sh
+mkdir -p control_logs/server-scale
+HEADSCALE_STRESS_NODES=500 GOMAXPROCS=4 \
+  go test ./hscontrol/servertest -run '^TestScale/500_nodes/personal_devices$' \
+  -count=1 -v -timeout=20m \
+  -cpuprofile=control_logs/server-scale/cpu.prof \
+  -memprofile=control_logs/server-scale/heap.prof \
+  >control_logs/server-scale/run.log 2>&1
+go tool pprof control_logs/server-scale/cpu.prof
+```
+
+Use `-trace=control_logs/server-scale/trace.out` in a separate diagnostic run
+to inspect scheduling, blocking, and GC. Profiles include setup, teardown,
+test clients, and assertions. In the pinned Tailscale dependency, zstd also
+performs extra runtime stack inspection in test binaries; confirm CPU
+findings against a standalone server before optimizing them. Docker/real
+tailscaled tests remain necessary for network behavior, data-plane traffic,
+and a server-only resource ceiling.

hscontrol/servertest/client.go

@@ -41,9 +41,11 @@ type TestClient struct {
 	pollDone   chan struct{}
 
 	// Accumulated state from [tailcfg.MapResponse] callbacks.
-	mu      sync.RWMutex
-	netmap  *netmap.NetworkMap
-	history []*netmap.NetworkMap
+	mu              sync.RWMutex
+	netmap          *netmap.NetworkMap
+	history         []*netmap.NetworkMap
+	updatesReceived int
+	noHistory       bool
 
 	// deltaUpdates and deltaRemoved: see [WithDeltaUpdates].
 	deltaUpdates bool
@@ -67,6 +69,14 @@ type clientConfig struct {
 	tags         []string
 	user         *types.User
 	deltaUpdates bool
+	noHistory    bool
+}
+
+// WithoutNetmapHistory retains only the latest netmap and one queued update.
+// Scale tests use this to avoid retaining every client's old peer lists.
+// [TestClient.UpdateCount] still counts every received full netmap.
+func WithoutNetmapHistory() ClientOption {
+	return func(c *clientConfig) { c.noHistory = true }
 }
 
 // WithDeltaUpdates makes the client accept incremental map updates the way
@@ -162,18 +172,24 @@ func NewClient(tb testing.TB, server *TestServer, name string, opts ...ClientOpt
 		tb.Fatalf("servertest: NewDirect(%s): %v", name, err)
 	}
 
+	updateBuffer := 64
+	if cc.noHistory {
+		updateBuffer = 1
+	}
+
 	tc := &TestClient{
 		Name:    name,
 		server:  server,
 		direct:  direct,
 		authKey: authKey,
 		user:    user,
-		updates: make(chan *netmap.NetworkMap, 64),
+		updates: make(chan *netmap.NetworkMap, updateBuffer),
 		bus:     bus,
 		dialer:  dialer,
 		tracker: tracker,
 
 		deltaUpdates: cc.deltaUpdates,
+		noHistory:    cc.noHistory,
 	}
 
 	tb.Cleanup(func() {
@@ -254,7 +270,11 @@ func (c *TestClient) resetNetmapState() {
 func (c *TestClient) UpdateFullNetmap(nm *netmap.NetworkMap) {
 	c.mu.Lock()
 	c.netmap = nm
-	c.history = append(c.history, nm)
+	c.updatesReceived++
+
+	if !c.noHistory {
+		c.history = append(c.history, nm)
+	}
 	c.mu.Unlock()
 
 	// Non-blocking send to the updates channel.
@@ -580,10 +600,11 @@ func (c *TestClient) UpdateCount() int {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
-	return len(c.history)
+	return c.updatesReceived
 }
 
-// History returns a copy of all [netmap.NetworkMap] snapshots in order.
+// History returns the retained [netmap.NetworkMap] snapshots in order.
+// It is empty when [WithoutNetmapHistory] is set.
 func (c *TestClient) History() []*netmap.NetworkMap {
 	c.mu.RLock()
 	defer c.mu.RUnlock()

hscontrol/servertest/scale_test.go

@@ -0,0 +1,557 @@
+package servertest
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"runtime"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/juanfont/headscale/hscontrol/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go4.org/netipx"
+	"tailscale.com/tailcfg"
+	"tailscale.com/tailcfg/nodecap"
+	"tailscale.com/types/ipproto"
+	"tailscale.com/types/logger"
+	"tailscale.com/types/netmap"
+	"tailscale.com/wgengine/filter"
+)
+
+const scaleDevicesPerUser = 10
+
+type scaleCase struct {
+	name          string
+	policy        string
+	taggedService bool
+	ssh           bool
+	peerCount     func(node, total int) int
+	canSend       func(source, destination int) bool
+}
+
+// TestScale runs the same workloads at smoke-test and explicitly selected sizes.
+// See README.md for large runs, resource constraints, and profiling commands.
+func TestScale(t *testing.T) {
+	nodes := 12
+
+	if value, ok := os.LookupEnv("HEADSCALE_STRESS_NODES"); ok {
+		var err error
+
+		nodes, err = strconv.Atoi(value)
+		require.NoError(t, err, "HEADSCALE_STRESS_NODES must be an integer")
+		require.GreaterOrEqual(t, nodes, 2, "need at least two connected nodes")
+	}
+
+	personalPeers := func(node, total int) int {
+		first := node / scaleDevicesPerUser * scaleDevicesPerUser
+		return min(scaleDevicesPerUser, total-first) - 1
+	}
+	personalVisibility := func(node, peer int) bool {
+		return node/scaleDevicesPerUser == peer/scaleDevicesPerUser
+	}
+
+	cases := []scaleCase{
+		{
+			name: "allow_all",
+			policy: `{
+				"grants": [{"src": ["*"], "dst": ["*"], "ip": ["tcp:443"]}]
+			}`,
+			peerCount: func(_, total int) int { return total - 1 },
+			canSend:   func(_, _ int) bool { return true },
+		},
+		{
+			name:      "isolated",
+			policy:    `{"acls": []}`,
+			peerCount: func(_, _ int) int { return 0 },
+			canSend:   func(_, _ int) bool { return false },
+		},
+		{
+			name: "personal_devices",
+			policy: `{
+				"grants": [{
+					"src": ["autogroup:member"], "dst": ["autogroup:self"],
+					"ip": ["tcp:443"]
+				}]
+			}`,
+			peerCount: personalPeers,
+			canSend:   personalVisibility,
+		},
+		{
+			name: "shared_service",
+			policy: `{
+				"tagOwners": {"tag:service": ["scale-user-0@"]},
+				"grants": [{
+					"src": ["autogroup:member"], "dst": ["tag:service"],
+					"ip": ["tcp:443"]
+				}]
+			}`,
+			taggedService: true,
+			peerCount: func(node, total int) int {
+				if node == 0 {
+					return total - 1
+				}
+
+				return 1
+			},
+			canSend: func(source, destination int) bool { return source != 0 && destination == 0 },
+		},
+		{
+			name: "personal_ssh",
+			policy: `{
+				"grants": [{
+					"src": ["autogroup:member"], "dst": ["autogroup:self"],
+					"ip": ["tcp:22", "tcp:443"]
+				}],
+				"ssh": [{
+					"action": "accept", "src": ["autogroup:member"],
+					"dst": ["autogroup:self"], "users": ["root"]
+				}]
+			}`,
+			ssh:       true,
+			peerCount: personalPeers,
+			canSend:   personalVisibility,
+		},
+	}
+
+	t.Run(fmt.Sprintf("%d_nodes", nodes), func(t *testing.T) {
+		// Run serially: each server and all its clients share this process.
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) { runScale(t, nodes, tc) })
+		}
+	})
+}
+
+// Discard client protocol chatter while keeping test failures and phase reports.
+type quietScaleTB struct{ testing.TB }
+
+func (quietScaleTB) Logf(string, ...any) {}
+
+func runScale(t *testing.T, nodes int, tc scaleCase) {
+	t.Helper()
+
+	workers := min(4, runtime.GOMAXPROCS(0))
+
+	opts := []ServerOption{WithBatcherWorkers(workers)}
+	if _, largeRun := os.LookupEnv("HEADSCALE_STRESS_NODES"); !largeRun {
+		opts = append(opts, WithNodeStoreBatchTimeout(10*time.Millisecond))
+	}
+
+	srv := NewServer(t, opts...)
+	clients := make([]*TestClient, nodes)
+	selfNodes := make([]tailcfg.NodeView, nodes)
+	indices := make(map[tailcfg.NodeID]int, nodes)
+
+	const convergenceTimeout = 2 * time.Minute
+
+	t.Logf(
+		"target_nodes=%d devices_per_user=%d workers=%d gomaxprocs=%d gomemlimit=%q go=%s",
+		nodes,
+		scaleDevicesPerUser,
+		workers,
+		runtime.GOMAXPROCS(0),
+		os.Getenv("GOMEMLIMIT"),
+		runtime.Version(),
+	)
+
+	phase := func(name string, work func()) {
+		var before, after runtime.MemStats
+		runtime.ReadMemStats(&before)
+
+		started := time.Now()
+		defer func() {
+			elapsed := time.Since(started)
+
+			runtime.ReadMemStats(&after)
+
+			peers := 0
+			live := 0
+
+			for _, c := range clients {
+				if c != nil {
+					peers += len(c.Peers())
+					select {
+					case <-c.pollDone:
+					default:
+						live++
+					}
+				}
+			}
+
+			t.Logf(
+				"phase=%s elapsed=%s allocated_bytes=%d heap_bytes=%d gc_cycles=%d goroutines=%d registered_nodes=%d live_clients=%d peer_entries=%d",
+				name,
+				elapsed,
+				after.TotalAlloc-before.TotalAlloc,
+				after.HeapAlloc,
+				after.NumGC-before.NumGC,
+				runtime.NumGoroutine(),
+				srv.State().ListNodes().Len(),
+				live,
+				peers,
+			)
+		}()
+
+		work()
+		checkScaleStreams(t, srv, clients, indices)
+	}
+
+	// A generation attribute distinguishes fresh policy delivery from old maps.
+	setPolicy := func(policy, generation string) {
+		var document map[string]any
+		require.NoError(t, json.Unmarshal([]byte(policy), &document))
+		document["nodeAttrs"] = []any{map[string]any{
+			"target": []string{"*"}, "attr": []string{generation},
+		}}
+		data, err := json.Marshal(document)
+		require.NoError(t, err)
+		_, err = srv.State().SetPolicyInDB(string(data))
+		require.NoError(t, err)
+		changes, err := srv.State().ReloadPolicy()
+		srv.App.Change(changes...)
+		require.NoError(t, err)
+	}
+
+	waitForState := func(generation string, revoked bool, osMarker string) {
+		deadline := time.Now().Add(convergenceTimeout)
+
+		for i, c := range clients {
+			wantPeers := tc.peerCount(i, nodes)
+			if revoked {
+				wantPeers = 0
+			}
+
+			c.WaitForCondition(
+				t,
+				fmt.Sprintf("%s: %d peers, OS=%q", generation, wantPeers, osMarker),
+				time.Until(deadline),
+				func(nm *netmap.NetworkMap) bool {
+					if !nm.SelfNode.CapMap().Contains(nodecap.Cap(generation)) ||
+						len(nm.Peers) != wantPeers {
+						return false
+					}
+
+					hasSSH := nm.SSHPolicy != nil && len(nm.SSHPolicy.Rules) > 0
+					if hasSSH != (tc.ssh && !revoked) {
+						return false
+					}
+
+					if revoked && nm.PacketFilterRules.Len() != 0 {
+						return false
+					}
+
+					return tc.peersMatch(nm, i, indices, osMarker)
+				},
+			)
+			checkScalePacketFilter(t, tc, c.Netmap(), selfNodes, i, revoked)
+
+			if tc.ssh && !revoked {
+				checkScaleSSH(t, c.Netmap().SSHPolicy, selfNodes, i)
+			}
+		}
+	}
+
+	phase("enroll", func() {
+		users := make([]*types.User, (nodes+scaleDevicesPerUser-1)/scaleDevicesPerUser)
+		for i := range users {
+			users[i] = srv.CreateUser(t, fmt.Sprintf("scale-user-%d", i))
+		}
+
+		setPolicy(tc.policy, "example.com/scale-initial")
+
+		// Bound concurrent logins, and join every batch before checking results.
+		for first := 0; first < nodes; first += 32 {
+			var registrations sync.WaitGroup
+			for i := first; i < min(first+32, nodes); i++ {
+				registrations.Go(func() {
+					opts := []ClientOption{
+						WithUser(users[i/scaleDevicesPerUser]), WithoutNetmapHistory(),
+					}
+					if tc.taggedService && i == 0 {
+						opts = append(opts, WithTags("tag:service"))
+					}
+
+					clients[i] = NewClient(quietScaleTB{t}, srv,
+						fmt.Sprintf("scale-node-%05d", i), opts...)
+				})
+			}
+
+			registrations.Wait()
+
+			if t.Failed() {
+				t.FailNow()
+			}
+		}
+
+		deadline := time.Now().Add(convergenceTimeout)
+		for i, c := range clients {
+			c.WaitForCondition(t, "initial map", time.Until(deadline),
+				func(nm *netmap.NetworkMap) bool { return nm.SelfNode.Valid() })
+			id := c.Netmap().SelfNode.ID()
+			selfNodes[i] = c.Netmap().SelfNode
+			_, duplicate := indices[id]
+			require.False(t, duplicate, "duplicate client node ID %d", id)
+			indices[id] = i
+		}
+
+		require.Equal(t, nodes, srv.State().ListNodes().Len())
+		waitForState("example.com/scale-initial", false, "")
+	})
+
+	// One sender per user, including the shared service, changes in each burst.
+	for round := range 3 {
+		marker := fmt.Sprintf("scale-os-%d", round)
+		phase(fmt.Sprintf("update_burst_%d", round), func() {
+			updateScaleClients(t, srv, clients, marker)
+			waitForState("example.com/scale-initial", false, marker)
+		})
+	}
+
+	phase("revoke_policy", func() {
+		setPolicy(`{"acls": []}`, "example.com/scale-revoked")
+		waitForState("example.com/scale-revoked", true, "")
+	})
+	phase("restore_policy", func() {
+		setPolicy(tc.policy, "example.com/scale-restored")
+		waitForState("example.com/scale-restored", false, "scale-os-2")
+	})
+	phase("reconnect_burst", func() {
+		var reconnects sync.WaitGroup
+		for i := 0; i < nodes; i += scaleDevicesPerUser {
+			reconnects.Go(func() { clients[i].Reconnect(quietScaleTB{t}) })
+		}
+
+		reconnects.Wait()
+
+		if t.Failed() {
+			t.FailNow()
+		}
+
+		waitForState("example.com/scale-restored", false, "scale-os-2")
+		setPolicy(tc.policy, "example.com/scale-reconnected")
+		waitForState("example.com/scale-reconnected", false, "scale-os-2")
+	})
+}
+
+func (tc scaleCase) peersMatch(
+	nm *netmap.NetworkMap,
+	viewer int,
+	indices map[tailcfg.NodeID]int,
+	osMarker string,
+) bool {
+	var previous tailcfg.NodeID
+
+	for _, peer := range nm.Peers {
+		j, ok := indices[peer.ID()]
+		if !ok || j == viewer || (!tc.canSend(viewer, j) && !tc.canSend(j, viewer)) ||
+			peer.ID() <= previous {
+			return false
+		}
+
+		previous = peer.ID()
+
+		online, known := peer.Online().GetOk()
+		if !known || !online {
+			return false
+		}
+
+		if osMarker != "" && j%scaleDevicesPerUser == 0 &&
+			peer.Hostinfo().OS() != osMarker {
+			return false
+		}
+	}
+
+	return true
+}
+
+func updateScaleClients(
+	t *testing.T,
+	srv *TestServer,
+	clients []*TestClient,
+	marker string,
+) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Minute)
+	defer cancel()
+
+	var senders sync.WaitGroup
+
+	for i := 0; i < len(clients); i += scaleDevicesPerUser {
+		c := clients[i]
+
+		senders.Go(func() {
+			c.Direct().SetHostinfo(&tailcfg.Hostinfo{
+				Hostname: c.Name, BackendLogID: "servertest-" + c.Name, OS: marker,
+			})
+			assert.NoError(t, c.Direct().SendUpdate(ctx), c.Name)
+		})
+	}
+
+	senders.Wait()
+
+	if t.Failed() {
+		t.FailNow()
+	}
+
+	for i := 0; i < len(clients); i += scaleDevicesPerUser {
+		node, ok := srv.State().GetNodeByNodeKey(clients[i].Netmap().NodeKey)
+		require.True(t, ok)
+		require.Equal(t, marker, node.Hostinfo().OS(), clients[i].Name)
+	}
+}
+
+// Probe a few sources per node instead of making the test itself all-pairs.
+func checkScalePacketFilter(
+	t *testing.T,
+	tc scaleCase,
+	nm *netmap.NetworkMap,
+	selfNodes []tailcfg.NodeView,
+	viewer int,
+	revoked bool,
+) {
+	t.Helper()
+
+	var localIPs netipx.IPSetBuilder
+	for _, prefix := range nm.SelfNode.Addresses().All() {
+		localIPs.AddPrefix(prefix)
+	}
+
+	local, err := localIPs.IPSet()
+	require.NoError(t, err)
+
+	packetFilter := filter.New(nm.PacketFilter, nil, local, nil, nil, logger.Discard)
+
+	first := viewer / scaleDevicesPerUser * scaleDevicesPerUser
+	groupSize := min(scaleDevicesPerUser, len(selfNodes)-first)
+
+	sibling := first + (viewer-first+1)%groupSize
+	for _, source := range []int{sibling, (first + scaleDevicesPerUser) % len(selfNodes), 0} {
+		for family, dst := range nm.SelfNode.Addresses().All() {
+			src := selfNodes[source].Addresses().At(family).Addr()
+
+			want := filter.Drop
+			if !revoked && tc.canSend(source, viewer) {
+				want = filter.Accept
+			}
+
+			require.Equal(t, want, packetFilter.CheckTCP(src, dst.Addr(), 443),
+				"%d -> %d TCP 443", selfNodes[source].ID(), selfNodes[viewer].ID())
+			require.Equal(t, filter.Drop, packetFilter.CheckTCP(src, dst.Addr(), 80))
+			require.Equal(
+				t,
+				filter.Drop,
+				packetFilter.Check(src, dst.Addr(), 443, ipproto.UDP),
+			)
+
+			if tc.ssh {
+				require.Equal(t, want, packetFilter.CheckTCP(src, dst.Addr(), 22))
+			}
+		}
+	}
+}
+
+func checkScaleSSH(
+	t *testing.T,
+	policy *tailcfg.SSHPolicy,
+	selfNodes []tailcfg.NodeView,
+	viewer int,
+) {
+	t.Helper()
+
+	require.NotNil(t, policy)
+	require.Len(t, policy.Rules, 1)
+	rule := policy.Rules[0]
+	require.NotNil(t, rule.Action)
+	require.True(t, rule.Action.Accept)
+	require.False(t, rule.Action.Reject)
+	require.Empty(t, rule.Action.HoldAndDelegate)
+	require.Equal(t, map[string]string{"root": "root"}, rule.SSHUsers)
+
+	wantIPs := make(map[string]bool)
+
+	first := viewer / scaleDevicesPerUser * scaleDevicesPerUser
+	for _, node := range selfNodes[first:min(first+scaleDevicesPerUser, len(selfNodes))] {
+		for _, prefix := range node.Addresses().All() {
+			wantIPs[prefix.Addr().String()] = true
+		}
+	}
+
+	require.Len(t, rule.Principals, len(wantIPs))
+
+	for _, principal := range rule.Principals {
+		require.False(t, principal.Any)
+		require.Empty(t, principal.Node)
+		require.Empty(t, principal.UserLogin)
+		require.True(
+			t,
+			wantIPs[principal.NodeIP],
+			"unexpected SSH source %s",
+			principal.NodeIP,
+		)
+		delete(wantIPs, principal.NodeIP)
+	}
+
+	require.Empty(t, wantIPs)
+}
+
+func checkScaleStreams(
+	t *testing.T,
+	srv *TestServer,
+	clients []*TestClient,
+	indices map[tailcfg.NodeID]int,
+) {
+	t.Helper()
+
+	require.Zero(t, srv.App.MapBatcher().WorkErrors(), "mapper work failed")
+	connections := srv.App.MapBatcher().Debug()
+	require.Len(t, connections, len(clients))
+
+	for id, info := range connections {
+		_, known := indices[id.NodeID()]
+		require.True(t, known, "unexpected connected node %s", id)
+		require.True(t, info.Connected, "node %s disconnected on server", id)
+		require.Equal(t, 1, info.ActiveConnections, "node %s", id)
+	}
+
+	for _, c := range clients {
+		require.NotNil(t, c, "client registration failed")
+
+		select {
+		case <-c.pollDone:
+			t.Fatalf("%s: control stream exited", c.Name)
+		default:
+		}
+
+		require.NotNil(t, c.Netmap(), c.Name)
+		require.Empty(t, c.History(), "scale clients must not retain netmaps")
+		require.Positive(t, c.UpdateCount(), c.Name)
+	}
+}
+
+func TestClientNetmapHistory(t *testing.T) {
+	for _, noHistory := range []bool{false, true} {
+		t.Run(fmt.Sprintf("disabled=%v", noHistory), func(t *testing.T) {
+			c := &TestClient{
+				noHistory: noHistory,
+				updates:   make(chan *netmap.NetworkMap, 1),
+			}
+			for range 3 {
+				c.UpdateFullNetmap(&netmap.NetworkMap{})
+			}
+
+			require.Equal(t, 3, c.UpdateCount())
+			require.NotNil(t, c.Netmap())
+
+			if noHistory {
+				require.Empty(t, c.History())
+			} else {
+				require.Len(t, c.History(), 3)
+			}
+		})
+	}
+}

hscontrol/servertest/server.go

@@ -40,17 +40,18 @@ type TestServer struct {
 type ServerOption func(*serverConfig)
 
 type serverConfig struct {
-	batchDelay       time.Duration
-	bufferedChanSize int
-	ephemeralTimeout time.Duration
-	nodeExpiry       time.Duration
-	batcherWorkers   int
-	taildropEnabled  bool
-	logTailEnabled   bool
-	realListener     bool
-	magicDNSDomain   string
-	dnsResolvers     []string
-	certProvider     hsdns.Provider
+	batchDelay            time.Duration
+	bufferedChanSize      int
+	ephemeralTimeout      time.Duration
+	nodeExpiry            time.Duration
+	batcherWorkers        int
+	nodeStoreBatchTimeout time.Duration
+	taildropEnabled       bool
+	logTailEnabled        bool
+	realListener          bool
+	magicDNSDomain        string
+	dnsResolvers          []string
+	certProvider          hsdns.Provider
 }
 
 func defaultServerConfig() *serverConfig {
@@ -80,6 +81,11 @@ func WithBatcherWorkers(n int) ServerOption {
 	return func(c *serverConfig) { c.batcherWorkers = n }
 }
 
+// WithNodeStoreBatchTimeout overrides the server's default write-batch wait.
+func WithNodeStoreBatchTimeout(d time.Duration) ServerOption {
+	return func(c *serverConfig) { c.nodeStoreBatchTimeout = d }
+}
+
 // WithEphemeralTimeout sets the ephemeral node inactivity timeout.
 func WithEphemeralTimeout(d time.Duration) ServerOption {
 	return func(c *serverConfig) { c.ephemeralTimeout = d }
@@ -176,6 +182,7 @@ func NewServer(tb testing.TB, opts ...ServerOption) *TestServer {
 			BatchChangeDelay:               sc.batchDelay,
 			BatcherWorkers:                 sc.batcherWorkers,
 			NodeMapSessionBufferedChanSize: sc.bufferedChanSize,
+			NodeStoreBatchTimeout:          sc.nodeStoreBatchTimeout,
 		},
 	}