From 7cd5c1732bb5374f21005073937c42f4d531e3c5 Mon Sep 17 00:00:00 2001 From: Theodor Midtlien Date: Wed, 8 Jul 2026 14:36:42 +0200 Subject: [PATCH] [client] Fix hanging status command during relay dial (#6694) * Add regression test for relay state lock * Make connect not hold a lock in openConnVia --- shared/relay/client/manager.go | 66 +++++++++----- shared/relay/client/manager_cleanup_test.go | 60 ++++++++++++ .../relay/client/manager_relaystates_test.go | 91 +++++++++++++++++++ 3 files changed, 196 insertions(+), 21 deletions(-) create mode 100644 shared/relay/client/manager_cleanup_test.go create mode 100644 shared/relay/client/manager_relaystates_test.go diff --git a/shared/relay/client/manager.go b/shared/relay/client/manager.go index e1515401e..2f2839d94 100644 --- a/shared/relay/client/manager.go +++ b/shared/relay/client/manager.go @@ -30,11 +30,16 @@ type RelayTrack struct { relayClient *Client err error created time.Time + // ready is closed once the dial started by openConnVia finishes (relayClient + // or err is set). Callers reusing a track wait on this instead of the track + // lock, so the dial never runs under rt.Lock. + ready chan struct{} } func NewRelayTrack() *RelayTrack { return &RelayTrack{ created: time.Now(), + ready: make(chan struct{}), } } @@ -326,34 +331,24 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string // check if already has a connection to the desired relay server m.relayClientsMutex.RLock() rt, ok := m.relayClients[serverAddress] - if ok { - rt.RLock() - m.relayClientsMutex.RUnlock() - defer rt.RUnlock() - if rt.err != nil { - return nil, rt.err - } - return rt.relayClient.OpenConn(ctx, peerKey) - } m.relayClientsMutex.RUnlock() + if ok { + return m.openConnOnTrack(ctx, rt, peerKey) + } // if not, establish a new connection but check it again (because changed the lock type) before starting the // connection m.relayClientsMutex.Lock() rt, ok = m.relayClients[serverAddress] if ok { - rt.RLock() m.relayClientsMutex.Unlock() - defer rt.RUnlock() - if rt.err != nil { - return nil, rt.err - } - return rt.relayClient.OpenConn(ctx, peerKey) + return m.openConnOnTrack(ctx, rt, peerKey) } - // create a new relay client and store it in the relayClients map + // Publish the track and release the map lock BEFORE dialing, so the dial does + // not run under rt.Lock (which would block RelayStates and the cleanup loop + // for the full dial). Concurrent callers find this track and wait on rt.ready. rt = NewRelayTrack() - rt.Lock() m.relayClients[serverAddress] = rt m.relayClientsMutex.Unlock() @@ -361,8 +356,10 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string relayClient.SetTransportFallback(m.transportFallback) err := relayClient.Connect(m.ctx) if err != nil { + rt.Lock() rt.err = err rt.Unlock() + close(rt.ready) m.relayClientsMutex.Lock() delete(m.relayClients, serverAddress) m.relayClientsMutex.Unlock() @@ -370,14 +367,34 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string } // if connection closed then delete the relay client from the list relayClient.SetOnDisconnectListener(m.onServerDisconnected) + rt.Lock() rt.relayClient = relayClient rt.Unlock() + close(rt.ready) - conn, err := relayClient.OpenConn(ctx, peerKey) - if err != nil { - return nil, err + return relayClient.OpenConn(ctx, peerKey) +} + +// openConnOnTrack opens a peer connection through an existing relay track, +// waiting for the dial started by another openConnVia call to finish. It waits +// on rt.ready rather than the track lock, so it neither holds nor contends the +// track lock across the dial. +func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) { + select { + case <-rt.ready: + case <-ctx.Done(): + return nil, ctx.Err() } - return conn, nil + + rt.RLock() + defer rt.RUnlock() + if rt.err != nil { + return nil, rt.err + } + if rt.relayClient == nil { + return nil, ErrRelayClientNotConnected + } + return rt.relayClient.OpenConn(ctx, peerKey) } func (m *Manager) onServerConnected() { @@ -476,6 +493,13 @@ func (m *Manager) cleanUpUnusedRelays() { continue } + // dial still in progress (openConnVia publishes the track before Connect + // completes and no longer holds rt.Lock during it), nothing to clean up. + if rt.relayClient == nil { + rt.Unlock() + continue + } + if time.Since(rt.created) <= m.keepUnusedServerTime { rt.Unlock() continue diff --git a/shared/relay/client/manager_cleanup_test.go b/shared/relay/client/manager_cleanup_test.go new file mode 100644 index 000000000..6ac5daeac --- /dev/null +++ b/shared/relay/client/manager_cleanup_test.go @@ -0,0 +1,60 @@ +package client + +import ( + "context" + "net/netip" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial drives a real, hanging foreign +// relay dial and asserts cleanUpUnusedRelays does not stall behind it. +func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) { + serverAddr := stallingRelayListener(t) + + mCtx, mCancel := context.WithCancel(context.Background()) + t.Cleanup(mCancel) + + m := NewManager(mCtx, nil, "alice", 1280) + + dialDone := make(chan struct{}) + go func() { + defer close(dialDone) + _, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{}) + }() + + // The track appears in the map once the dial is in flight. + require.Eventually(t, func() bool { + m.relayClientsMutex.RLock() + defer m.relayClientsMutex.RUnlock() + _, ok := m.relayClients[serverAddr] + return ok + }, 5*time.Second, 5*time.Millisecond, "relay dial did not start") + + cleanupDone := make(chan struct{}) + go func() { + defer close(cleanupDone) + m.cleanUpUnusedRelays() + }() + + select { + case <-cleanupDone: + case <-time.After(2 * time.Second): + t.Fatal("cleanUpUnusedRelays blocked on an in-progress relay dial while holding the relay map lock") + } + + m.relayClientsMutex.RLock() + _, stillTracked := m.relayClients[serverAddr] + m.relayClientsMutex.RUnlock() + require.True(t, stillTracked, "an in-progress relay dial must not be evicted by cleanup") + + // Release the hanging dial so the goroutine can exit cleanly. + mCancel() + select { + case <-dialDone: + case <-time.After(5 * time.Second): + t.Fatal("openConnVia did not return after context cancellation") + } +} diff --git a/shared/relay/client/manager_relaystates_test.go b/shared/relay/client/manager_relaystates_test.go new file mode 100644 index 000000000..f26323323 --- /dev/null +++ b/shared/relay/client/manager_relaystates_test.go @@ -0,0 +1,91 @@ +package client + +import ( + "context" + "net" + "net/netip" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// stallingRelayListener accepts TCP connections and holds them open without ever +// responding, so a relay handshake dialed against it blocks until its context is +// cancelled. It returns the "rel://host:port" URL to dial. +func stallingRelayListener(t *testing.T) string { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + var mu sync.Mutex + var conns []net.Conn + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + mu.Lock() + conns = append(conns, c) + mu.Unlock() + } + }() + t.Cleanup(func() { + _ = ln.Close() + mu.Lock() + for _, c := range conns { + _ = c.Close() + } + mu.Unlock() + }) + + return "rel://" + ln.Addr().String() +} + +// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for +// RelayStates() called by a "status -d command" hanging behind an in-progress +// relay dial. +func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) { + serverAddr := stallingRelayListener(t) + + mCtx, mCancel := context.WithCancel(context.Background()) + t.Cleanup(mCancel) + + m := NewManager(mCtx, nil, "alice", 1280) + + dialDone := make(chan struct{}) + go func() { + defer close(dialDone) + _, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{}) + }() + + require.Eventually(t, func() bool { + m.relayClientsMutex.RLock() + defer m.relayClientsMutex.RUnlock() + _, ok := m.relayClients[serverAddr] + return ok + }, 5*time.Second, 5*time.Millisecond, "relay dial did not start") + + done := make(chan []RelayConnState, 1) + go func() { + done <- m.RelayStates() + }() + + select { + case states := <-done: + require.Empty(t, states, "a relay still being dialed carries no state and must be omitted") + case <-time.After(2 * time.Second): + t.Fatal("RelayStates blocked on a foreign relay whose Connect() is in progress") + } + + // Release the hanging dial so the goroutine can exit cleanly. + mCancel() + select { + case <-dialDone: + case <-time.After(5 * time.Second): + t.Fatal("openConnVia did not return after context cancellation") + } +}