mapper: consolidate builder and connection helpers

This commit is contained in:
Kristoffer Dalby
2026-06-16 11:33:35 +00:00
parent 03e2d24c79
commit 3979887d8d
4 changed files with 47 additions and 34 deletions
+2 -10
View File
@@ -315,11 +315,7 @@ func (b *Batcher) AddNode(
initialMap, err := b.MapResponseFromChange(id, change.FullSelf(id)) initialMap, err := b.MapResponseFromChange(id, change.FullSelf(id))
if err != nil { if err != nil {
nlog.Error().Err(err).Msg("initial map generation failed") nlog.Error().Err(err).Msg("initial map generation failed")
nodeConn.removeConnectionByChannel(c) nodeConn.detach(c)
if !nodeConn.hasActiveConnections() {
nodeConn.markDisconnected()
}
return fmt.Errorf("generating initial map for node %d: %w", id, err) return fmt.Errorf("generating initial map for node %d: %w", id, err)
} }
@@ -343,11 +339,7 @@ func (b *Batcher) AddNode(
nlog.Error().Err(ErrInitialMapSendTimeout).Msg("initial map send timeout") nlog.Error().Err(ErrInitialMapSendTimeout).Msg("initial map send timeout")
nlog.Debug().Caller().Dur("timeout.duration", 5*time.Second). //nolint:mnd nlog.Debug().Caller().Dur("timeout.duration", 5*time.Second). //nolint:mnd
Msg("initial map send timed out because channel was blocked or receiver not ready") Msg("initial map send timed out because channel was blocked or receiver not ready")
nodeConn.removeConnectionByChannel(c) nodeConn.detach(c)
if !nodeConn.hasActiveConnections() {
nodeConn.markDisconnected()
}
return fmt.Errorf("%w for node %d", ErrInitialMapSendTimeout, id) return fmt.Errorf("%w for node %d", ErrInitialMapSendTimeout, id)
} }
+15 -10
View File
@@ -61,6 +61,16 @@ func (b *MapResponseBuilder) hasErrors() bool {
return len(b.errs) > 0 return len(b.errs) > 0
} }
// node looks up the requesting node, recording ErrNodeNotFoundMapper on miss.
func (b *MapResponseBuilder) node() (types.NodeView, bool) {
nv, ok := b.mapper.state.GetNodeByID(b.nodeID)
if !ok {
b.addError(ErrNodeNotFoundMapper)
}
return nv, ok
}
// WithCapabilityVersion sets the capability version for the response. // WithCapabilityVersion sets the capability version for the response.
func (b *MapResponseBuilder) WithCapabilityVersion(capVer tailcfg.CapabilityVersion) *MapResponseBuilder { func (b *MapResponseBuilder) WithCapabilityVersion(capVer tailcfg.CapabilityVersion) *MapResponseBuilder {
b.capVer = capVer b.capVer = capVer
@@ -69,9 +79,8 @@ func (b *MapResponseBuilder) WithCapabilityVersion(capVer tailcfg.CapabilityVers
// WithSelfNode adds the requesting node to the response. // WithSelfNode adds the requesting node to the response.
func (b *MapResponseBuilder) WithSelfNode() *MapResponseBuilder { func (b *MapResponseBuilder) WithSelfNode() *MapResponseBuilder {
nv, ok := b.mapper.state.GetNodeByID(b.nodeID) nv, ok := b.node()
if !ok { if !ok {
b.addError(ErrNodeNotFoundMapper)
return b return b
} }
@@ -136,9 +145,8 @@ func (b *MapResponseBuilder) WithDebugConfig() *MapResponseBuilder {
// WithSSHPolicy adds SSH policy configuration for the requesting node. // WithSSHPolicy adds SSH policy configuration for the requesting node.
func (b *MapResponseBuilder) WithSSHPolicy() *MapResponseBuilder { func (b *MapResponseBuilder) WithSSHPolicy() *MapResponseBuilder {
node, ok := b.mapper.state.GetNodeByID(b.nodeID) node, ok := b.node()
if !ok { if !ok {
b.addError(ErrNodeNotFoundMapper)
return b return b
} }
@@ -155,9 +163,8 @@ func (b *MapResponseBuilder) WithSSHPolicy() *MapResponseBuilder {
// WithDNSConfig adds DNS configuration for the requesting node. // WithDNSConfig adds DNS configuration for the requesting node.
func (b *MapResponseBuilder) WithDNSConfig() *MapResponseBuilder { func (b *MapResponseBuilder) WithDNSConfig() *MapResponseBuilder {
node, ok := b.mapper.state.GetNodeByID(b.nodeID) node, ok := b.node()
if !ok { if !ok {
b.addError(ErrNodeNotFoundMapper)
return b return b
} }
@@ -168,9 +175,8 @@ func (b *MapResponseBuilder) WithDNSConfig() *MapResponseBuilder {
// WithUserProfiles adds user profiles for the requesting node and given peers. // WithUserProfiles adds user profiles for the requesting node and given peers.
func (b *MapResponseBuilder) WithUserProfiles(peers views.Slice[types.NodeView]) *MapResponseBuilder { func (b *MapResponseBuilder) WithUserProfiles(peers views.Slice[types.NodeView]) *MapResponseBuilder {
node, ok := b.mapper.state.GetNodeByID(b.nodeID) node, ok := b.node()
if !ok { if !ok {
b.addError(ErrNodeNotFoundMapper)
return b return b
} }
@@ -185,9 +191,8 @@ func (b *MapResponseBuilder) WithUserProfiles(peers views.Slice[types.NodeView])
// For autogroup:self policies, it returns per-node compiled rules. // For autogroup:self policies, it returns per-node compiled rules.
// For global policies, it returns the global filter reduced for this node. // For global policies, it returns the global filter reduced for this node.
func (b *MapResponseBuilder) WithPacketFilters() *MapResponseBuilder { func (b *MapResponseBuilder) WithPacketFilters() *MapResponseBuilder {
node, ok := b.mapper.state.GetNodeByID(b.nodeID) node, ok := b.node()
if !ok { if !ok {
b.addError(ErrNodeNotFoundMapper)
return b return b
} }
+20 -14
View File
@@ -499,15 +499,9 @@ func (m *mapper) filterVisiblePeerPatches(
return nil return nil
} }
var filtered []*tailcfg.PeerChange return filterByVisible(visible, patches, func(p *tailcfg.PeerChange) tailcfg.NodeID {
return p.NodeID
for _, patch := range patches { })
if _, vis := visible[patch.NodeID]; vis {
filtered = append(filtered, patch)
}
}
return filtered
} }
// filterVisibleNodes restricts a peer slice to the nodes the recipient can see // filterVisibleNodes restricts a peer slice to the nodes the recipient can see
@@ -524,15 +518,27 @@ func (m *mapper) filterVisibleNodes(
return views.SliceOf([]types.NodeView{}) return views.SliceOf([]types.NodeView{})
} }
var filtered []types.NodeView return views.SliceOf(filterByVisible(visible, peers.AsSlice(), func(p types.NodeView) tailcfg.NodeID {
return p.ID().NodeID()
}))
}
for _, peer := range peers.All() { // filterByVisible keeps only the items whose key resolves to a NodeID present
if _, vis := visible[peer.ID().NodeID()]; vis { // in the visible set, preserving input order.
filtered = append(filtered, peer) func filterByVisible[T any](
visible map[tailcfg.NodeID]struct{},
items []T,
key func(T) tailcfg.NodeID,
) []T {
var filtered []T
for _, it := range items {
if _, ok := visible[key(it)]; ok {
filtered = append(filtered, it)
} }
} }
return views.SliceOf(filtered) return filtered
} }
func writeDebugMapResponse( func writeDebugMapResponse(
+10
View File
@@ -183,6 +183,16 @@ func (mc *multiChannelNodeConn) removeConnectionByChannel(c chan<- *tailcfg.MapR
return false return false
} }
// detach removes the connection for the given channel and marks the node
// disconnected if no active connections remain.
func (mc *multiChannelNodeConn) detach(c chan<- *tailcfg.MapResponse) {
mc.removeConnectionByChannel(c)
if !mc.hasActiveConnections() {
mc.markDisconnected()
}
}
// hasActiveConnections checks if the node has any active connections. // hasActiveConnections checks if the node has any active connections.
func (mc *multiChannelNodeConn) hasActiveConnections() bool { func (mc *multiChannelNodeConn) hasActiveConnections() bool {
mc.mutex.RLock() mc.mutex.RLock()