state, change: send a deleted node's removal only to its peers

The removal went to every node, also those the policy hid the node from.
The delete now returns the node's peers from the snapshot it replaces.

hscontrol/mapper/batch_coalesce_test.go

@@ -88,7 +88,8 @@ func TestAppendPendingStaysBoundedWhileInFlight(t *testing.T) {
 
 		cs := []change.Change{change.NodeAdded(peer), change.NodeOnline(peer), change.PolicyChange()}
 		if i%100 == 0 {
-			cs = append(cs, change.NodeRemoved(types.NodeID(i+1000))) //nolint:gosec // test
+			// The removal this node gets as a peer of the deleted one.
+			cs = append(cs, change.NodeRemoved(types.NodeID(i+1000), id)[1]) //nolint:gosec // test
 		}
 
 		for _, c := range cs {

hscontrol/mapper/batcher.go

@@ -514,8 +514,8 @@ func (b *Batcher) RemoveNode(id types.NodeID, c chan<- *tailcfg.MapResponse) boo
 
 // AddWork queues a change to be processed by the batcher. Empty changes
 // carry no work for any recipient and are dropped here so they never create
-// pending entries or in-flight work. A PeersRemoved change is not empty, so
-// deletion cleanup in addToBatch still runs.
+// pending entries or in-flight work. A deletion's teardown change is not
+// empty, so its cleanup in addToBatch still runs.
 func (b *Batcher) AddWork(r ...change.Change) {
 	if slices.ContainsFunc(r, change.Change.IsEmpty) {
 		kept := slices.DeleteFunc(slices.Clone(r), change.Change.IsEmpty)

hscontrol/mapper/batcher_concurrency_test.go

@@ -292,7 +292,7 @@ func TestAddToBatchDifferential(t *testing.T) {
 			change.PeersChanged("several peers", 3, 1),
 		}},
 		{"deletion", []change.Change{
-			change.NodeRemoved(4), change.SelfUpdate(4), change.PeersRemoved(3),
+			change.NodeRemoved(4, 1)[1], change.SelfUpdate(4), change.PeersRemoved(3),
 		}},
 		{
 			"full_with_pings",
@@ -511,7 +511,7 @@ func TestAddToBatch_NodeRemovalCleanup(t *testing.T) {
 	_, exists := lb.b.nodes.Load(removedNode)
 	require.True(t, exists, "node 3 should exist before removal")
 
-	lb.b.addToBatch(change.NodeRemoved(removedNode))
+	lb.b.addToBatch(change.NodeRemoved(removedNode)...)
 
 	// Node should be removed from the nodes map
 	_, exists = lb.b.nodes.Load(removedNode)

hscontrol/mapper/batcher_test.go

@@ -2758,26 +2758,114 @@ func TestSSHPolicyEmptyOnWire(t *testing.T) {
 	}
 }
 
-// TestAddWorkPeersRemovedNotTreatedAsEmpty ensures the empty filter does
-// not swallow a PeersRemoved change — deletion cleanup must still run, and
-// surviving recipients must still see the removal.
-func TestAddWorkPeersRemovedNotTreatedAsEmpty(t *testing.T) {
+// TestAddWorkDeletionNotTreatedAsEmpty ensures the empty filter does not
+// swallow a deletion — its teardown must still run, and the deleted node's
+// peers must still see the removal.
+func TestAddWorkDeletionNotTreatedAsEmpty(t *testing.T) {
 	lb := setupLightweightBatcher(t, 4, 16)
 	defer lb.cleanup()
 
-	// Remove node 4.
-	removal := change.NodeRemoved(types.NodeID(4))
-	require.False(t, removal.IsEmpty(), "PeersRemoved change is not empty")
+	// Remove node 4, a peer of nodes 1 and 2 only.
+	removal := change.NodeRemoved(types.NodeID(4), 1, 2)
+	require.False(t, removal[0].IsEmpty(), "the teardown change is not empty")
 
-	lb.b.AddWork(removal)
+	lb.b.AddWork(removal...)
 
 	// Node 4 must be evicted from the batcher.
 	_, exists := lb.b.nodes.Load(types.NodeID(4))
 	require.False(t, exists, "removed node must be evicted from batcher")
 
-	// Surviving nodes must see the removal pending.
-	require.Equal(t, 3, countNodesPending(lb.b),
-		"surviving 3 nodes must have the removal pending")
+	// Only its peers have the removal pending.
+	require.Equal(t, 2, countNodesPending(lb.b),
+		"only the deleted node's 2 peers must have the removal pending")
+}
+
+// TestDeleteWhileInitialMapInFlight deletes a peer while a node's initial
+// map, which still lists it, is in flight. The node must be told the peer is
+// gone once the map is out, also when the policy refresh fails and only the
+// deletion's own removals go out.
+func TestDeleteWhileInitialMapInFlight(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		policyChange bool
+	}{
+		{"with_policy_change", true},
+		{"removal_only", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			td, cleanup := setupBatcherWithTestData(
+				t,
+				NewBatcherAndMapper,
+				1,
+				3,
+				normalBufferSize,
+			)
+			defer cleanup()
+
+			b := td.Batcher
+			n1, n2, n3 := &td.Nodes[0], &td.Nodes[1], &td.Nodes[2]
+			capVer := tailcfg.CapabilityVersion(100)
+
+			n2.start()
+			n3.start()
+
+			defer n2.cleanup()
+			defer n3.cleanup()
+
+			require.NoError(t, b.AddNode(n2.n.ID, n2.ch, capVer, nil))
+			require.NoError(t, b.AddNode(n3.n.ID, n3.ch, capVer, nil))
+
+			// Unbuffered and unread: AddNode blocks on the initial map.
+			ch1 := make(chan *tailcfg.MapResponse)
+			addErr := make(chan error, 1)
+
+			go func() { addErr <- b.AddNode(n1.n.ID, ch1, capVer, nil) }()
+
+			waitInitialMapSendParked(t)
+
+			v3, ok := td.State.GetNodeByID(n3.n.ID)
+			require.True(t, ok)
+
+			changes, err := td.State.DeleteNode(v3)
+			require.NoError(t, err)
+
+			if !tc.policyChange {
+				changes = slices.DeleteFunc(changes, func(c change.Change) bool {
+					return c.RequiresRuntimePeerComputation
+				})
+			}
+
+			b.AddWork(changes...)
+
+			// Let ticks render n1's changes while its stream waits.
+			time.Sleep(300 * time.Millisecond) //nolint:forbidigo // test coordination
+
+			initial := <-ch1
+			require.NotNil(t, initial.Node, "first frame is the initial map")
+
+			initialPeers := make([]tailcfg.NodeID, 0, len(initial.Peers))
+			for _, p := range initial.Peers {
+				initialPeers = append(initialPeers, p.ID)
+			}
+
+			require.Contains(t, initialPeers, n3.n.ID.NodeID(),
+				"the initial map predates the delete")
+
+			deadline := time.After(3 * time.Second)
+
+			for {
+				select {
+				case resp := <-ch1:
+					if slices.Contains(resp.PeersRemoved, n3.n.ID.NodeID()) {
+						require.NoError(t, <-addErr)
+						return
+					}
+				case <-deadline:
+					t.Fatal("n1 was never told that deleted n3 is gone")
+				}
+			}
+		})
+	}
 }
 
 // TestHandleNodeChangeSendsRemovalsAsDelta checks that peers missing from a

hscontrol/mapper/batcher_unit_test.go

@@ -877,7 +877,7 @@ func TestAddToBatch_NodeRemovedStopsSession(t *testing.T) {
 	mc.connections[0].stop = func() { close(stopped) }
 	mc.mutex.Unlock()
 
-	lb.b.AddWork(change.NodeRemoved(1))
+	lb.b.AddWork(change.NodeRemoved(1)...)
 
 	// addToBatch runs synchronously, so the session must already be stopped.
 	select {

hscontrol/servertest/issues_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/netip"
+	"slices"
 	"testing"
 	"time"
 
@@ -1137,6 +1138,52 @@ func TestPeerRemovedAsDelta(t *testing.T) {
 	})
 }
 
+// TestDeletedNodeNotAnnouncedToHiddenPeers deletes a node another user's node
+// cannot see under the policy: only the deleted node's peers are told it is
+// gone.
+func TestDeletedNodeNotAnnouncedToHiddenPeers(t *testing.T) {
+	t.Parallel()
+
+	srv := servertest.NewServer(t)
+	_, err := srv.State().SetPolicy([]byte(`{
+		"acls": [
+			{"action": "accept", "src": ["alice@"], "dst": ["alice@:*"]},
+			{"action": "accept", "src": ["bob@"], "dst": ["bob@:*"]}
+		]
+	}`))
+	require.NoError(t, err)
+
+	alice := srv.CreateUser(t, "alice")
+	bob := srv.CreateUser(t, "bob")
+
+	laptop := servertest.NewClient(t, srv, "alice-laptop",
+		servertest.WithUser(alice), servertest.WithDeltaUpdates())
+	servertest.NewClient(t, srv, "alice-db", servertest.WithUser(alice))
+	other := servertest.NewClient(t, srv, "bob-node",
+		servertest.WithUser(bob), servertest.WithDeltaUpdates())
+
+	laptop.WaitForPeers(t, 1, 10*time.Second)
+	other.WaitForPeers(t, 0, 10*time.Second)
+
+	db := findNodeID(t, srv, "alice-db")
+	dbView, ok := srv.State().GetNodeByID(db)
+	require.True(t, ok)
+
+	changes, err := srv.State().DeleteNode(dbView)
+	require.NoError(t, err)
+	srv.App.Change(changes...)
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.Contains(c, laptop.DeltaRemovedPeers(), db.NodeID())
+	}, 10*time.Second, 50*time.Millisecond, "a peer of the deleted node must be told it is gone")
+
+	// bob-node's stream got the same batch; give it time to deliver.
+	assert.Never(t, func() bool {
+		return slices.Contains(other.DeltaRemovedPeers(), db.NodeID())
+	}, 2*time.Second, 50*time.Millisecond,
+		"a node the policy hid the deleted node from must not be told of it")
+}
+
 func findNodeID(tb testing.TB, srv *servertest.TestServer, hostname string) types.NodeID {
 	tb.Helper()
 

hscontrol/servertest/mapstream_compat_test.go

@@ -82,10 +82,6 @@ var mapStreamSkipReasons = map[string]string{
 	"map-delivery/S3.6/bystander/peers": sameUserPeers,
 	"map-delivery/S3.6/laptop/peers":    sameUserPeers,
 
-	"hidden-peers/S9/observer/removed":  removalSentToAll,
-	"hidden-peers/S10/observer/removed": removalSentToAll,
-	"hidden-peers/S13/observer/removed": removalSentToAll,
-
 	"hidden-peers/S2/witness/presence":   policyResponseToEveryNode,
 	"hidden-peers/S8/observer/presence":  policyResponseToEveryNode,
 	"hidden-peers/S9/observer/presence":  policyResponseToEveryNode,
@@ -129,7 +125,6 @@ const (
 	restartEndpointsKept      = "RESTART_ENDPOINTS_KEPT: headscale keeps a restarted node's last endpoints at its peers until it reports new ones; SaaS sends peers the node without endpoints until then"
 	reconnectAnnounced        = "RECONNECT_ANNOUNCED: headscale tells peers a node is online after a stream gap of a few seconds; SaaS sends them nothing"
 	routerReconnectRecompute  = "ROUTER_RECONNECT_RECOMPUTE: headscale sends every node a policy response when a subnet router's stream comes back after a few seconds; SaaS sends nothing"
-	removalSentToAll          = "REMOVAL_SENT_TO_ALL: headscale tells every connected node that a deleted node is gone; SaaS tells only the nodes that had it as a peer"
 	taggedDeviceUser          = "TAGGED_DEVICE_USER: SaaS shows a tagged device's user as the login that made its key; headscale shows tagged-devices"
 )
 

hscontrol/state/node_store.go

@@ -228,10 +228,12 @@ type PeersFunc func(nodes []types.NodeView) map[types.NodeID][]types.NodeID
 
 // work represents a single operation to be performed on the [NodeStore].
 type work struct {
-	op         int
-	nodeID     types.NodeID
-	node       types.Node
-	result     chan struct{}
+	op     int
+	nodeID types.NodeID
+	node   types.Node
+	result chan struct{}
+	// For del: receives the node's peers in the snapshot the delete replaces.
+	peers      *[]types.NodeID
 	nodeResult chan types.NodeView
 	// For rebuildPeerMaps operation
 	rebuildResult chan struct{}
@@ -397,16 +399,20 @@ func (s *NodeStore) updateNodes(
 	nodeStoreOperations.WithLabelValues("update_multi").Inc()
 }
 
-// DeleteNode removes a node from the store by its ID.
+// DeleteNode removes a node from the store by its ID and returns its peers
+// in the last snapshot that held it: every node that may have been sent it.
 // This is a blocking operation that waits for the write to complete.
-func (s *NodeStore) DeleteNode(id types.NodeID) {
+func (s *NodeStore) DeleteNode(id types.NodeID) []types.NodeID {
 	timer := prometheus.NewTimer(nodeStoreOperationDuration.WithLabelValues("delete"))
 	defer timer.ObserveDuration()
 
+	var peers []types.NodeID
+
 	work := work{
 		op:     del,
 		nodeID: id,
 		result: make(chan struct{}),
+		peers:  &peers,
 	}
 
 	nodeStoreQueueDepth.Inc()
@@ -416,13 +422,15 @@ func (s *NodeStore) DeleteNode(id types.NodeID) {
 	case <-s.stopped:
 		nodeStoreQueueDepth.Dec()
 
-		return
+		return nil
 	}
 
 	<-work.result
 	nodeStoreQueueDepth.Dec()
 
 	nodeStoreOperations.WithLabelValues("delete").Inc()
+
+	return peers
 }
 
 // SetGivenName sets [types.Node.GivenName] on the node identified by id,
@@ -542,8 +550,9 @@ func (s *NodeStore) applyBatch(batch []work) {
 
 	nodeStoreBatchSize.Observe(float64(len(batch)))
 
+	prev := s.data.Load()
 	nodes := make(map[types.NodeID]types.Node)
-	maps.Copy(nodes, s.data.Load().nodesByID)
+	maps.Copy(nodes, prev.nodesByID)
 
 	// Track which work items need node results
 	nodeResultRequests := make(map[types.NodeID][]*work)
@@ -605,6 +614,10 @@ func (s *NodeStore) applyBatch(batch []work) {
 			}
 		case del:
 			delete(nodes, w.nodeID)
+
+			if w.peers != nil {
+				*w.peers = slices.Clone(prev.peersByNode[w.nodeID])
+			}
 			// For delete operations, send an invalid NodeView if requested
 			if w.nodeResult != nil {
 				nodeResultRequests[w.nodeID] = append(nodeResultRequests[w.nodeID], w)
@@ -649,8 +662,6 @@ func (s *NodeStore) applyBatch(batch []work) {
 		}
 	}
 
-	prev := s.data.Load()
-
 	// A relation change recomputes adjacency; anything else reuses the
 	// previous peer IDs and re-elects routes only if an election input moved.
 	if relationChanged {

hscontrol/state/node_store_test.go

@@ -2729,3 +2729,27 @@ func TestNodeStoreNodeKeyIndexMatchesNodes(t *testing.T) {
 
 	require.Empty(t, nodeKeyIndexMismatch(store.data.Load()))
 }
+
+// TestDeleteNodeReturnsPeersOfReplacedSnapshot pins that a delete reports the
+// node's peers in the snapshot it replaces: every node that may have been sent
+// it, read in the same write as the delete.
+func TestDeleteNodeReturnsPeersOfReplacedSnapshot(t *testing.T) {
+	nodes := types.Nodes{}
+
+	for id := types.NodeID(1); id <= 5; id++ {
+		n := createTestNode(id, 1, "user", fmt.Sprintf("node%d", id))
+		nodes = append(nodes, &n)
+	}
+
+	store := NewNodeStore(nodes, oddEvenPeersFunc, TestBatchSize, TestBatchTimeout)
+
+	store.Start()
+	defer store.Stop()
+
+	peers := store.DeleteNode(3)
+	slices.Sort(peers)
+	assert.Equal(t, []types.NodeID{1, 5}, peers, "odd node 3 is seen only by odd nodes")
+
+	assert.Equal(t, []types.NodeID{4}, store.DeleteNode(2))
+	assert.Empty(t, store.DeleteNode(42), "an unknown node has no peers")
+}

hscontrol/state/persist_test.go

@@ -792,7 +792,19 @@ func TestDeleteNodeKeepsStoreOnDBFailure(t *testing.T) {
 }
 
 func TestDeleteNodeReturnsRemovalOnPolicyFailure(t *testing.T) {
-	_, s, nodeID := persistTestSetup(t)
+	cfg := persistTestConfig(t.TempDir() + "/headscale.db")
+
+	database, err := db.NewHeadscaleDatabase(cfg)
+	require.NoError(t, err)
+
+	user := database.CreateUserForTest("persist-user")
+	nodeID := database.CreateRegisteredNodeForTest(user, "persist-node").ID
+	peerID := database.CreateRegisteredNodeForTest(user, "persist-peer").ID
+
+	require.NoError(t, database.Close())
+
+	s, err := NewState(cfg)
+	require.NoError(t, err)
 	t.Cleanup(func() { _ = s.Close() })
 
 	node, ok := s.GetNodeByID(nodeID)
@@ -812,11 +824,13 @@ func TestDeleteNodeReturnsRemovalOnPolicyFailure(t *testing.T) {
 
 	cs, err := s.DeleteNode(node)
 	require.ErrorIs(t, err, errInjectedPolicyNodeUpdate)
-	require.NotEmpty(t, cs)
-	assert.Equal(t, []types.NodeID{nodeID}, cs[0].PeersRemoved,
-		"a committed deletion must still notify peers and stop the node's session")
+	require.GreaterOrEqual(t, len(cs), 2)
 	assert.Equal(t, []types.NodeID{nodeID}, cs[0].DeletedNodes,
 		"a committed deletion must identify the session to stop")
+	assert.Equal(t, peerID, cs[1].TargetNode,
+		"a committed deletion must still notify peers")
+	assert.Equal(t, []types.NodeID{nodeID}, cs[1].PeersRemoved,
+		"a committed deletion must still notify peers")
 
 	_, ok = s.GetNodeByID(nodeID)
 	assert.False(t, ok, "a committed deletion must remove the in-memory node")

hscontrol/state/state.go

@@ -743,20 +743,22 @@ func (s *State) DeleteNode(node types.NodeView) ([]change.Change, error) {
 	// node look deleted until the next restart. persistMu stays out of this
 	// write: a persist that still finds the node updates no row, and holding
 	// it would give every concurrent delete a NodeStore batch of its own.
-	s.nodeStore.DeleteNode(node.ID())
+	peers := s.nodeStore.DeleteNode(node.ID())
 
 	s.ipAlloc.FreeIPs(node.IPs())
 
-	// An explicit removal of its own, ahead of the policy refresh, so peers
-	// learn of the deletion without depending on their sent-peers tracking.
-	removed := change.NodeRemoved(node.ID())
+	// Explicit removals of their own, ahead of the policy refresh, so peers
+	// learn of the deletion without depending on their sent-peers tracking,
+	// and nodes the policy hid it from learn nothing.
+	changes := change.NodeRemoved(node.ID(), peers...)
 
 	policyChange, err := s.updatePolicyManagerNodes(genBefore)
 	if err != nil {
-		return []change.Change{removed, policyChange}, fmt.Errorf("updating policy manager after node deletion: %w", err)
+		return append(changes, policyChange),
+			fmt.Errorf("updating policy manager after node deletion: %w", err)
 	}
 
-	return []change.Change{removed, policyChange}, nil
+	return append(changes, policyChange), nil
 }
 
 // Connect acquires a control session and returns the resulting changes

hscontrol/types/change/change.go

@@ -562,13 +562,28 @@ func NodeAdded(id types.NodeID) Change {
 	return c
 }
 
-// NodeRemoved returns a [Change] for when a node is removed.
-func NodeRemoved(id types.NodeID) Change {
-	return Change{
+// NodeRemoved returns the changes for a deleted node: the teardown of its own
+// sessions, which reaches no other node, and its removal from each of peers.
+// Each names the node in DeletedNodes, so a peer's tick folds it with the
+// tick's other deletions, see [CoalescePeerDeltas].
+func NodeRemoved(id types.NodeID, peers ...types.NodeID) []Change {
+	out := make([]Change, 0, 1+len(peers))
+	out = append(out, Change{
 		Reason:       "node removed",
-		PeersRemoved: []types.NodeID{id},
+		TargetNode:   id,
 		DeletedNodes: []types.NodeID{id},
+	})
+
+	for _, peer := range peers {
+		out = append(out, Change{
+			Reason:       "node removed",
+			TargetNode:   peer,
+			PeersRemoved: []types.NodeID{id},
+			DeletedNodes: []types.NodeID{id},
+		})
 	}
+
+	return out
 }
 
 // KeyExpiryFor returns a [Change] for when a node's key expiry changes.

hscontrol/types/change/change_test.go

@@ -529,7 +529,7 @@ func TestCoalescePeerDeltas(t *testing.T) {
 		{
 			name: "deletions fold into one removal ahead of the tick",
 			changes: []Change{
-				NodeAdded(2), NodeRemoved(5), NodeOnline(3), PolicyChange(), NodeRemoved(4), NodeAdded(6),
+				NodeAdded(2), NodeRemoved(5, self)[1], NodeOnline(3), PolicyChange(), NodeRemoved(4, self)[1], NodeAdded(6),
 			},
 			want: []Change{
 				{Reason: "node removed", PeersRemoved: []types.NodeID{4, 5}, DeletedNodes: []types.NodeID{4, 5}},
@@ -622,7 +622,7 @@ func TestCoalescePeerDeltasHoistsOnlyDeletions(t *testing.T) {
 			want, ok := hoists[field.Name]
 			require.True(t, ok, "field %s is not classified", field.Name)
 
-			c := NodeRemoved(2)
+			c := NodeRemoved(2, self)[1]
 			setNonZero(t, reflect.ValueOf(&c).Elem().FieldByIndex(field.Index), self)
 
 			assert.Equal(t, want, c.isDeletion())
@@ -710,10 +710,21 @@ func TestPeersRemoved(t *testing.T) {
 }
 
 func TestNodeRemoved(t *testing.T) {
-	r := NodeRemoved(42)
-	assert.Equal(t, "node removed", r.Reason)
-	assert.Equal(t, []types.NodeID{42}, r.PeersRemoved)
-	assert.Equal(t, []types.NodeID{42}, r.DeletedNodes)
+	rs := NodeRemoved(42, 7, 9)
+	require.Len(t, rs, 3)
+	assert.Equal(t, "node removed", rs[0].Reason)
+	assert.Equal(t, types.NodeID(42), rs[0].TargetNode,
+		"the teardown reaches no other node")
+	assert.Empty(t, rs[0].PeersRemoved)
+	assert.Equal(t, []types.NodeID{42}, rs[0].DeletedNodes)
+
+	for i, peer := range []types.NodeID{7, 9} {
+		assert.Equal(t, peer, rs[i+1].TargetNode)
+		assert.Equal(t, []types.NodeID{42}, rs[i+1].PeersRemoved)
+		assert.Equal(t, []types.NodeID{42}, rs[i+1].DeletedNodes,
+			"a peer's removal folds with the tick's other deletions")
+		assert.True(t, rs[i+1].isDeletion())
+	}
 }
 
 func TestPeerPatched(t *testing.T) {