diff --git a/hscontrol/mapper/batcher.go b/hscontrol/mapper/batcher.go index 25811291..a820ccef 100644 --- a/hscontrol/mapper/batcher.go +++ b/hscontrol/mapper/batcher.go @@ -283,15 +283,28 @@ func (b *Batcher) AddNode( // Initialize last used timestamp newEntry.lastUsed.Store(now.Unix()) - // Get or create multiChannelNodeConn - this reuses existing offline nodes for rapid reconnection - nodeConn, loaded := b.nodes.LoadOrStore(id, newMultiChannelNodeConn(id, b.mapper)) + // Get or create the multiChannelNodeConn and register this connection in a + // single Compute so the new connection is visible atomically. Doing the + // LoadOrStore and addConnection as separate steps let cleanupOfflineNodes + // (which deletes via Compute when hasActiveConnections() is false) observe a + // reused offline conn with zero connections mid-reconnect and delete it, + // orphaning the live connection. + var nodeConn *multiChannelNodeConn - if !loaded { - b.totalNodes.Add(1) - } + b.nodes.Compute( + id, + func(existing *multiChannelNodeConn, loaded bool) (*multiChannelNodeConn, xsync.ComputeOp) { + if !loaded || existing == nil { + existing = newMultiChannelNodeConn(id, b.mapper) + b.totalNodes.Add(1) + } - // Add connection to the list (lock-free) - nodeConn.addConnection(newEntry) + existing.addConnection(newEntry) + nodeConn = existing + + return existing, xsync.UpdateOp + }, + ) // Use the worker pool for controlled concurrency instead of direct generation initialMap, err := b.MapResponseFromChange(id, change.FullSelf(id)) diff --git a/hscontrol/mapper/reconnect_cleanup_test.go b/hscontrol/mapper/reconnect_cleanup_test.go new file mode 100644 index 00000000..4d97a1e8 --- /dev/null +++ b/hscontrol/mapper/reconnect_cleanup_test.go @@ -0,0 +1,59 @@ +package mapper + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "tailscale.com/tailcfg" +) + +// TestAddNodeReconnectNotOrphanedByCleanup ensures a node reconnecting via +// AddNode is not deleted from b.nodes by a concurrent cleanupOfflineNodes pass. +// AddNode must register the connection atomically with the get-or-create, so +// the offline-cleanup Compute either sees the active connection (and cancels) +// or runs first and AddNode recreates the entry — never leaving a live +// connection orphaned outside b.nodes. +func TestAddNodeReconnectNotOrphanedByCleanup(t *testing.T) { + testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, 1, normalBufferSize) + defer cleanup() + + b := testData.Batcher.Batcher + node := &testData.Nodes[0] + + testData.State.Connect(node.n.ID) + + go func() { + for range node.ch { + } + }() + + require.NoError(t, b.AddNode(node.n.ID, node.ch, tailcfg.CapabilityVersion(100), nil)) + + // Model a long-offline conn awaiting cleanup or a rapid reconnect. + nc, ok := b.nodes.Load(node.n.ID) + require.True(t, ok) + + nc.removeConnectionByChannel(node.ch) + + past := time.Now().Add(-(offlineNodeCleanupThreshold + time.Minute)) + nc.disconnectedAt.Store(&past) + require.False(t, nc.hasActiveConnections()) + + var wg sync.WaitGroup + + wg.Go(func() { + _ = b.AddNode(node.n.ID, node.ch, tailcfg.CapabilityVersion(100), nil) + }) + + wg.Go(func() { + b.cleanupOfflineNodes() + }) + + wg.Wait() + + assert.True(t, b.IsConnected(node.n.ID), + "reconnecting node was orphaned: live connection absent from b.nodes") +}