From d4f2acf3ab3cada875ab88eab4e0f81e4118cd12 Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Wed, 1 Jul 2026 09:40:21 +0000 Subject: [PATCH] policy: take RLock for reads so map generation runs concurrently One exclusive mutex serialized every policy read, so a mass reconnect on autogroup:self/via/relay policies stalled clients into "unexpected EOF" retries. Per-node caches become xsync.Maps for lazy population under RLock. Fixes #3346 (cherry picked from commit 6f317c7576474c5d19c799f4e225d66d2b597e50) --- CHANGELOG.md | 8 ++ hscontrol/policy/v2/policy.go | 136 ++++++++++++++++------------- hscontrol/policy/v2/policy_test.go | 20 +++-- hscontrol/policy/v2/sshtest.go | 4 +- hscontrol/policy/v2/test.go | 4 +- 5 files changed, 98 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6630e6c3..406f65d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ **Minimum supported Tailscale client version: v1.xx.0** +## 0.29.2 (202x-xx-xx) + +**Minimum supported Tailscale client version: v1.80.0** + +### Changes + +- Fix map generation serializing on the policy lock, so a mass reconnect on `autogroup:self`, via or relay policies no longer stalls clients into `unexpected EOF` retry loops [#3358](https://github.com/juanfont/headscale/pull/3358) + ## 0.29.1 (2026-06-18) **Minimum supported Tailscale client version: v1.80.0** diff --git a/hscontrol/policy/v2/policy.go b/hscontrol/policy/v2/policy.go index a7b55a2a..a98d0191 100644 --- a/hscontrol/policy/v2/policy.go +++ b/hscontrol/policy/v2/policy.go @@ -15,6 +15,7 @@ import ( "github.com/juanfont/headscale/hscontrol/policy/matcher" "github.com/juanfont/headscale/hscontrol/policy/policyutil" "github.com/juanfont/headscale/hscontrol/types" + "github.com/puzpuzpuz/xsync/v4" "github.com/rs/zerolog/log" "go4.org/netipx" "tailscale.com/net/tsaddr" @@ -28,7 +29,10 @@ import ( var ErrInvalidTagOwner = errors.New("tag owner is not an Alias") type PolicyManager struct { - mu sync.Mutex + // RWMutex, not Mutex, so concurrent map generation does not serialise on + // reads. The per-node caches are xsync.Maps so a read can fill them without + // taking the write lock. + mu sync.RWMutex pol *Policy users []types.User nodes views.Slice[types.NodeView] @@ -55,7 +59,7 @@ type PolicyManager struct { viaTargetTags map[Tag]struct{} // Lazy map of SSH policies - sshPolicyMap map[types.NodeID]*tailcfg.SSHPolicy + sshPolicyMap *xsync.Map[types.NodeID, *tailcfg.SSHPolicy] // compiledGrants are the grants with sources pre-resolved. // The single source of truth for filter compilation. Both @@ -64,12 +68,12 @@ type PolicyManager struct { userNodeIdx userNodeIndex // Lazy map of per-node filter rules (reduced, for packet filters) - filterRulesMap map[types.NodeID][]tailcfg.FilterRule + filterRulesMap *xsync.Map[types.NodeID, []tailcfg.FilterRule] // Lazy map of per-node matchers derived from UNREDUCED filter // rules. Only populated on the slow path when needsPerNodeFilter // is true; the fast path returns pm.matchers directly. - matchersForNodeMap map[types.NodeID][]matcher.Match + matchersForNodeMap *xsync.Map[types.NodeID, []matcher.Match] // needsPerNodeFilter is true when any compiled grant requires // per-node work (autogroup:self or via grants). @@ -197,9 +201,9 @@ func NewPolicyManager(b []byte, users []types.User, nodes views.Slice[types.Node pol: policy, users: users, nodes: nodes, - sshPolicyMap: make(map[types.NodeID]*tailcfg.SSHPolicy, nodes.Len()), - filterRulesMap: make(map[types.NodeID][]tailcfg.FilterRule, nodes.Len()), - matchersForNodeMap: make(map[types.NodeID][]matcher.Match, nodes.Len()), + sshPolicyMap: xsync.NewMap[types.NodeID, *tailcfg.SSHPolicy](), + filterRulesMap: xsync.NewMap[types.NodeID, []tailcfg.FilterRule](), + matchersForNodeMap: xsync.NewMap[types.NodeID, []matcher.Match](), } _, err = pm.updateLocked() @@ -354,9 +358,9 @@ func (pm *PolicyManager) updateLocked() (bool, error) { // TODO(kradalby): This could potentially be optimized by only clearing the // policies for nodes that have changed. Particularly if the only difference is // that nodes has been added or removed. - clear(pm.sshPolicyMap) - clear(pm.filterRulesMap) - clear(pm.matchersForNodeMap) + pm.sshPolicyMap.Clear() + pm.filterRulesMap.Clear() + pm.matchersForNodeMap.Clear() } // If nothing changed, no need to update nodes @@ -400,8 +404,8 @@ func (pm *PolicyManager) NodeNeedsPeerRecompute(node types.NodeView) bool { return true } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() if pm.relayTargetIPs != nil && node.InIPSet(pm.relayTargetIPs) { return true @@ -422,10 +426,10 @@ func (pm *PolicyManager) NodeNeedsPeerRecompute(node types.NodeView) bool { // /machine/ssh/action/{src}/to/{dst}?local_user={local_user} per the // SaaS wire format. Cache is invalidated on policy reload. func (pm *PolicyManager) SSHPolicy(baseURL string, node types.NodeView) (*tailcfg.SSHPolicy, error) { - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() - if sshPol, ok := pm.sshPolicyMap[node.ID()]; ok { + if sshPol, ok := pm.sshPolicyMap.Load(node.ID()); ok { return sshPol, nil } @@ -434,7 +438,7 @@ func (pm *PolicyManager) SSHPolicy(baseURL string, node types.NodeView) (*tailcf return nil, fmt.Errorf("compiling SSH policy: %w", err) } - pm.sshPolicyMap[node.ID()] = sshPol + pm.sshPolicyMap.Store(node.ID(), sshPol) return sshPol, nil } @@ -450,8 +454,8 @@ func (pm *PolicyManager) SSHPolicy(baseURL string, node types.NodeView) (*tailcf func (pm *PolicyManager) SSHCheckParams( srcNodeID, dstNodeID types.NodeID, ) (time.Duration, bool) { - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() if pm.pol == nil || len(pm.pol.SSHs) == 0 { return 0, false @@ -584,8 +588,8 @@ func (pm *PolicyManager) Filter() ([]tailcfg.FilterRule, []matcher.Match) { return nil, nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() return pm.filter, pm.matchers } @@ -604,8 +608,8 @@ func (pm *PolicyManager) BuildPeerMap(nodes views.Slice[types.NodeView]) map[typ return nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // Precompute each node's subnet routes and exit-node status once; the // O(n^2) pair scans below would otherwise recompute them for every pair. @@ -726,7 +730,7 @@ func (pm *PolicyManager) filterForNodeLocked( return nil } - if rules, ok := pm.filterRulesMap[node.ID()]; ok { + if rules, ok := pm.filterRulesMap.Load(node.ID()); ok { return rules } @@ -738,7 +742,7 @@ func (pm *PolicyManager) filterForNodeLocked( } reduced := policyutil.ReduceFilterRules(node, unreduced) - pm.filterRulesMap[node.ID()] = reduced + pm.filterRulesMap.Store(node.ID(), reduced) return reduced } @@ -755,8 +759,8 @@ func (pm *PolicyManager) FilterForNode(node types.NodeView) ([]tailcfg.FilterRul return nil, nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() return pm.filterForNodeLocked(node), nil } @@ -776,8 +780,8 @@ func (pm *PolicyManager) MatchersForNode(node types.NodeView) ([]matcher.Match, return nil, nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // For global policies, return the shared global matchers. // Via grants require per-node matchers because the global matchers @@ -786,7 +790,7 @@ func (pm *PolicyManager) MatchersForNode(node types.NodeView) ([]matcher.Match, return pm.matchers, nil } - if cached, ok := pm.matchersForNodeMap[node.ID()]; ok { + if cached, ok := pm.matchersForNodeMap.Load(node.ID()); ok { return cached, nil } @@ -794,7 +798,7 @@ func (pm *PolicyManager) MatchersForNode(node types.NodeView) ([]matcher.Match, // the stored compiled grants for this specific node. unreduced := pm.filterRulesForNodeLocked(node) matchers := matcher.MatchesFromFilterRules(unreduced) - pm.matchersForNodeMap[node.ID()] = matchers + pm.matchersForNodeMap.Store(node.ID(), matchers) return matchers, nil } @@ -813,7 +817,7 @@ func (pm *PolicyManager) SetUsers(users []types.User) (bool, error) { // Clear SSH policy map when users change to force SSH policy recomputation // This ensures that if SSH policy compilation previously failed due to missing users, // it will be retried with the new user list - clear(pm.sshPolicyMap) + pm.sshPolicyMap.Clear() changed, err := pm.updateLocked() if err != nil { @@ -866,9 +870,9 @@ func (pm *PolicyManager) SetNodes(nodes views.Slice[types.NodeView]) (bool, erro if !needsUpdate { // This ensures fresh filter rules are generated for all nodes - clear(pm.sshPolicyMap) - clear(pm.filterRulesMap) - clear(pm.matchersForNodeMap) + pm.sshPolicyMap.Clear() + pm.filterRulesMap.Clear() + pm.matchersForNodeMap.Clear() } // Always return true when nodes changed, even if filter hash didn't change // (can happen with autogroup:self or when nodes are added but don't affect rules) @@ -921,8 +925,8 @@ func (pm *PolicyManager) NodeCanHaveTag(node types.NodeView, tag string) bool { return false } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // pm.pol is written by SetPolicy under pm.mu; reading it before the // lock races with concurrent policy reloads. @@ -1010,8 +1014,8 @@ func (pm *PolicyManager) TagExists(tag string) bool { return false } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // pm.pol is written by SetPolicy under pm.mu; reading it before the // lock races with concurrent policy reloads. @@ -1029,8 +1033,8 @@ func (pm *PolicyManager) NodeCanApproveRoute(node types.NodeView, route netip.Pr return false } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // If the route to-be-approved is an exit route, then we need to check // if the node is in allowed to approve it. This is treated differently @@ -1092,8 +1096,8 @@ func (pm *PolicyManager) ViaRoutesForPeer(viewer, peer types.NodeView) types.Via return result } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() // pm.pol is written by SetPolicy under pm.mu; reading it before the // lock races with concurrent policy reloads. @@ -1311,8 +1315,8 @@ func (pm *PolicyManager) DebugString() string { // pm.pol, filter, matchers, and the derived maps are all written // under pm.mu by SetPolicy/SetUsers/SetNodes. - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() var sb strings.Builder @@ -1475,7 +1479,7 @@ func (pm *PolicyManager) invalidateAutogroupSelfCache(oldNodes, newNodes views.S // Clear cache entries for affected users only. // For autogroup:self, we need to clear all nodes belonging to affected users // because autogroup:self rules depend on the entire user's device set. - for nodeID := range pm.filterRulesMap { + pm.filterRulesMap.Range(func(nodeID types.NodeID, _ []tailcfg.FilterRule) bool { // Find the user for this cached node var nodeUserID types.UserID @@ -1518,20 +1522,22 @@ func (pm *PolicyManager) invalidateAutogroupSelfCache(oldNodes, newNodes views.S // If we found the user and they're affected, clear this cache entry if found { if _, affected := affectedUsers[nodeUserID]; affected { - delete(pm.filterRulesMap, nodeID) - delete(pm.matchersForNodeMap, nodeID) + pm.filterRulesMap.Delete(nodeID) + pm.matchersForNodeMap.Delete(nodeID) } } else { // Node not found in either old or new list, clear it - delete(pm.filterRulesMap, nodeID) - delete(pm.matchersForNodeMap, nodeID) + pm.filterRulesMap.Delete(nodeID) + pm.matchersForNodeMap.Delete(nodeID) } - } + + return true + }) if len(affectedUsers) > 0 { log.Debug(). Int("affected_users", len(affectedUsers)). - Int("remaining_cache_entries", len(pm.filterRulesMap)). + Int("remaining_cache_entries", pm.filterRulesMap.Size()). Msg("Selectively cleared autogroup:self cache for affected users") } } @@ -1572,23 +1578,27 @@ func (pm *PolicyManager) invalidateGlobalPolicyCache(newNodes views.Slice[types. } if newNode.HasNetworkChanges(oldNode) { - delete(pm.filterRulesMap, nodeID) - delete(pm.matchersForNodeMap, nodeID) + pm.filterRulesMap.Delete(nodeID) + pm.matchersForNodeMap.Delete(nodeID) } } // Remove deleted nodes from cache - for nodeID := range pm.filterRulesMap { + pm.filterRulesMap.Range(func(nodeID types.NodeID, _ []tailcfg.FilterRule) bool { if _, exists := newNodeMap[nodeID]; !exists { - delete(pm.filterRulesMap, nodeID) + pm.filterRulesMap.Delete(nodeID) } - } - for nodeID := range pm.matchersForNodeMap { + return true + }) + + pm.matchersForNodeMap.Range(func(nodeID types.NodeID, _ []matcher.Match) bool { if _, exists := newNodeMap[nodeID]; !exists { - delete(pm.matchersForNodeMap, nodeID) + pm.matchersForNodeMap.Delete(nodeID) } - } + + return true + }) } // flattenTags resolves nested tag-owner references. Cycles @@ -1770,8 +1780,8 @@ func (pm *PolicyManager) NodeCapMap(id types.NodeID) tailcfg.NodeCapMap { return nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() src := pm.nodeAttrsMap[id] if len(src) == 0 { @@ -1794,8 +1804,8 @@ func (pm *PolicyManager) NodeCapMaps() map[types.NodeID]tailcfg.NodeCapMap { return nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() out := make(map[types.NodeID]tailcfg.NodeCapMap, len(pm.nodeAttrsMap)) maps.Copy(out, pm.nodeAttrsMap) diff --git a/hscontrol/policy/v2/policy_test.go b/hscontrol/policy/v2/policy_test.go index 9237f33f..43febe1a 100644 --- a/hscontrol/policy/v2/policy_test.go +++ b/hscontrol/policy/v2/policy_test.go @@ -8,6 +8,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/juanfont/headscale/hscontrol/policy/matcher" "github.com/juanfont/headscale/hscontrol/types" + "github.com/puzpuzpuz/xsync/v4" "github.com/stretchr/testify/require" "gorm.io/gorm" "tailscale.com/net/tsaddr" @@ -122,7 +123,7 @@ func TestInvalidateAutogroupSelfCache(t *testing.T) { require.NoError(t, err) } - require.Len(t, pm.filterRulesMap, len(initialNodes)) + require.Equal(t, len(initialNodes), pm.filterRulesMap.Size()) tests := []struct { name string @@ -207,19 +208,20 @@ func TestInvalidateAutogroupSelfCache(t *testing.T) { } } - pm.filterRulesMap = make(map[types.NodeID][]tailcfg.FilterRule) + pm.filterRulesMap.Clear() + for _, n := range initialNodes { _, err := pm.FilterForNode(n.View()) require.NoError(t, err) } - initialCacheSize := len(pm.filterRulesMap) + initialCacheSize := pm.filterRulesMap.Size() require.Equal(t, len(initialNodes), initialCacheSize) pm.invalidateAutogroupSelfCache(initialNodes.ViewSlice(), tt.newNodes.ViewSlice()) // Verify the expected number of cache entries were cleared - finalCacheSize := len(pm.filterRulesMap) + finalCacheSize := pm.filterRulesMap.Size() clearedEntries := initialCacheSize - finalCacheSize require.Equal(t, tt.expectedCleared, clearedEntries, tt.description) }) @@ -498,15 +500,19 @@ func TestInvalidateGlobalPolicyCache(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { pm := &PolicyManager{ - nodes: tt.oldNodes.ViewSlice(), - filterRulesMap: tt.initialCache, + nodes: tt.oldNodes.ViewSlice(), + filterRulesMap: xsync.NewMap[types.NodeID, []tailcfg.FilterRule](), + matchersForNodeMap: xsync.NewMap[types.NodeID, []matcher.Match](), + } + for id, rules := range tt.initialCache { + pm.filterRulesMap.Store(id, rules) } pm.invalidateGlobalPolicyCache(tt.newNodes.ViewSlice()) // Verify cache state for nodeID, shouldExist := range tt.expectedCacheAfter { - _, exists := pm.filterRulesMap[nodeID] + _, exists := pm.filterRulesMap.Load(nodeID) require.Equal(t, shouldExist, exists, "node %d cache existence mismatch", nodeID) } }) diff --git a/hscontrol/policy/v2/sshtest.go b/hscontrol/policy/v2/sshtest.go index d2b07a63..48089f7c 100644 --- a/hscontrol/policy/v2/sshtest.go +++ b/hscontrol/policy/v2/sshtest.go @@ -128,8 +128,8 @@ func (pm *PolicyManager) RunSSHTests() error { return nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() cache := make(map[types.NodeID]*tailcfg.SSHPolicy) results := runSSHPolicyTests(pm.pol, pm.users, pm.nodes, cache) diff --git a/hscontrol/policy/v2/test.go b/hscontrol/policy/v2/test.go index 5f1189f2..e28c2ef6 100644 --- a/hscontrol/policy/v2/test.go +++ b/hscontrol/policy/v2/test.go @@ -204,8 +204,8 @@ func (pm *PolicyManager) RunTests() error { return nil } - pm.mu.Lock() - defer pm.mu.Unlock() + pm.mu.RLock() + defer pm.mu.RUnlock() results := runPolicyTests(pm.pol, pm.filter, pm.users, pm.nodes) if results.AllPassed {