mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-22 08:21:30 +02:00
Compare commits
10 Commits
main
...
refactor/r
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8a46353ae2 | ||
|
|
d663da8f82 | ||
|
|
cf101c44b4 | ||
|
|
79b51a79e4 | ||
|
|
a737504ec9 | ||
|
|
3b1beb3497 | ||
|
|
49c0aeb6ce | ||
|
|
6774a43eae | ||
|
|
50a29c07ce | ||
|
|
7d8e20030b |
@@ -54,19 +54,15 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
|
||||
w.relaySupportedOnRemotePeer.Store(true)
|
||||
|
||||
// the relayManager will return with error in case if the connection has lost with relay server
|
||||
currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
|
||||
_, _, err := w.relayManager.RelayInstanceAddress()
|
||||
if err != nil {
|
||||
w.log.Errorf("failed to handle new offer: %s", err)
|
||||
return
|
||||
}
|
||||
|
||||
srv := w.preferredRelayServer(currentRelayAddress, remoteOfferAnswer.RelaySrvAddress)
|
||||
var serverIP netip.Addr
|
||||
if srv == remoteOfferAnswer.RelaySrvAddress {
|
||||
serverIP = remoteOfferAnswer.RelaySrvIP
|
||||
}
|
||||
|
||||
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
|
||||
preferForeign := !w.isController
|
||||
remoteRelayServer := relayClient.RelayServer{Addr: remoteOfferAnswer.RelaySrvAddress, IP: remoteOfferAnswer.RelaySrvIP}
|
||||
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, remoteRelayServer, w.config.Key, preferForeign)
|
||||
if err != nil {
|
||||
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
|
||||
w.log.Debugf("handled offer by reusing existing relay connection")
|
||||
@@ -80,14 +76,13 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
|
||||
w.relayedConn = relayedConn
|
||||
w.relayLock.Unlock()
|
||||
|
||||
err = w.relayManager.AddCloseListener(srv, w.onRelayClientDisconnected)
|
||||
if err != nil {
|
||||
log.Errorf("failed to add close listener: %s", err)
|
||||
if err := w.relayManager.AddCloseListener(relayedConn.RemoteAddr().String(), w.onRelayClientDisconnected); err != nil {
|
||||
w.log.Errorf("failed to add close listener: %s", err)
|
||||
_ = relayedConn.Close()
|
||||
return
|
||||
}
|
||||
|
||||
w.log.Debugf("peer conn opened via Relay: %s", srv)
|
||||
w.log.Debugf("peer conn opened via Relay: %s", relayedConn.RemoteAddr())
|
||||
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
|
||||
relayedConn: relayedConn,
|
||||
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
|
||||
@@ -126,13 +121,6 @@ func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
|
||||
return answer.RelaySrvAddress != ""
|
||||
}
|
||||
|
||||
func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress string) string {
|
||||
if w.isController {
|
||||
return myRelayAddress
|
||||
}
|
||||
return remoteRelayAddress
|
||||
}
|
||||
|
||||
func (w *WorkerRelay) onRelayClientDisconnected() {
|
||||
go w.conn.onRelayDisconnected()
|
||||
}
|
||||
|
||||
176
shared/relay/client/fallback_opener.go
Normal file
176
shared/relay/client/fallback_opener.go
Normal file
@@ -0,0 +1,176 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
raceTotalTimeout = 40 * time.Second
|
||||
raceFallbackDelay = 10 * time.Second
|
||||
)
|
||||
|
||||
type raceAttempt struct {
|
||||
conn net.Conn
|
||||
err error
|
||||
}
|
||||
|
||||
type raceOutcome struct {
|
||||
conn net.Conn
|
||||
err error
|
||||
done bool
|
||||
}
|
||||
|
||||
type connRace struct {
|
||||
opener *FallbackOpener
|
||||
peerKey string
|
||||
remoteRelayServer RelayServer
|
||||
preferForeign bool
|
||||
|
||||
raceCtx context.Context
|
||||
otherCtx context.Context
|
||||
cancelPreferred context.CancelFunc
|
||||
cancelOther context.CancelFunc
|
||||
results chan raceAttempt
|
||||
fallbackTimer *time.Timer
|
||||
|
||||
otherStarted bool
|
||||
settled int
|
||||
lastErr error
|
||||
}
|
||||
|
||||
type FallbackOpener struct {
|
||||
home *Client
|
||||
foreignStore *ForeignRelaysStore
|
||||
|
||||
fallbackDelay time.Duration
|
||||
totalTimeout time.Duration
|
||||
// openFn performs a single attempt. It is overridable in tests; when nil the
|
||||
// real home/foreign dispatch in open is used.
|
||||
openFn func(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt
|
||||
}
|
||||
|
||||
func NewFallbackOpener(home *Client, foreignStore *ForeignRelaysStore) *FallbackOpener {
|
||||
return &FallbackOpener{
|
||||
home: home,
|
||||
foreignStore: foreignStore,
|
||||
fallbackDelay: raceFallbackDelay,
|
||||
totalTimeout: raceTotalTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *FallbackOpener) Run(ctx context.Context, peerKey string, remoteRelayServer RelayServer, preferForeign bool) (net.Conn, error) {
|
||||
raceCtx, cancel := context.WithTimeout(ctx, r.totalTimeout)
|
||||
defer cancel()
|
||||
|
||||
preferredCtx, cancelPreferred := context.WithCancel(raceCtx)
|
||||
otherCtx, cancelOther := context.WithCancel(raceCtx)
|
||||
|
||||
race := &connRace{
|
||||
opener: r,
|
||||
peerKey: peerKey,
|
||||
remoteRelayServer: remoteRelayServer,
|
||||
preferForeign: preferForeign,
|
||||
raceCtx: raceCtx,
|
||||
otherCtx: otherCtx,
|
||||
cancelPreferred: cancelPreferred,
|
||||
cancelOther: cancelOther,
|
||||
results: make(chan raceAttempt, 2),
|
||||
fallbackTimer: time.NewTimer(r.fallbackDelay),
|
||||
}
|
||||
defer race.fallbackTimer.Stop()
|
||||
|
||||
go func() {
|
||||
race.results <- r.open(preferredCtx, peerKey, remoteRelayServer, preferForeign)
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-race.fallbackTimer.C:
|
||||
race.startOther()
|
||||
case res := <-race.results:
|
||||
if o := race.handleResult(res); o.done {
|
||||
return o.conn, o.err
|
||||
}
|
||||
case <-raceCtx.Done():
|
||||
return race.onTimeout()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *connRace) startOther() {
|
||||
if c.otherStarted {
|
||||
return
|
||||
}
|
||||
c.otherStarted = true
|
||||
c.fallbackTimer.Stop()
|
||||
go func() {
|
||||
c.results <- c.opener.open(c.otherCtx, c.peerKey, c.remoteRelayServer, !c.preferForeign)
|
||||
}()
|
||||
}
|
||||
|
||||
func (c *connRace) handleResult(res raceAttempt) raceOutcome {
|
||||
if (res.err == nil && res.conn != nil) || errors.Is(res.err, ErrConnAlreadyExists) {
|
||||
c.settled++
|
||||
c.stop()
|
||||
return raceOutcome{conn: res.conn, err: res.err, done: true}
|
||||
}
|
||||
|
||||
c.lastErr = res.err
|
||||
c.settled++
|
||||
if !c.otherStarted {
|
||||
c.startOther()
|
||||
return raceOutcome{}
|
||||
}
|
||||
if c.settled == 2 {
|
||||
c.cancelPreferred()
|
||||
c.cancelOther()
|
||||
return raceOutcome{err: c.lastErr, done: true}
|
||||
}
|
||||
return raceOutcome{}
|
||||
}
|
||||
|
||||
func (c *connRace) onTimeout() (net.Conn, error) {
|
||||
c.stop()
|
||||
if c.lastErr != nil {
|
||||
return nil, c.lastErr
|
||||
}
|
||||
return nil, c.raceCtx.Err()
|
||||
}
|
||||
|
||||
func (c *connRace) stop() {
|
||||
c.cancelPreferred()
|
||||
c.cancelOther()
|
||||
go c.opener.drainLoser(c.results, c.settled, c.otherStarted)
|
||||
}
|
||||
|
||||
func (r *FallbackOpener) open(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt {
|
||||
if r.openFn != nil {
|
||||
return r.openFn(ctx, peerKey, remoteRelayServer, foreign)
|
||||
}
|
||||
if foreign {
|
||||
conn, err := r.foreignStore.OpenConn(ctx, peerKey, remoteRelayServer)
|
||||
return raceAttempt{conn: conn, err: err}
|
||||
}
|
||||
conn, err := r.home.OpenConn(ctx, peerKey)
|
||||
return raceAttempt{conn: conn, err: err}
|
||||
}
|
||||
|
||||
func (r *FallbackOpener) drainLoser(results chan raceAttempt, settled int, otherStarted bool) {
|
||||
started := 1
|
||||
if otherStarted {
|
||||
started = 2
|
||||
}
|
||||
for i := settled; i < started; i++ {
|
||||
res := <-results
|
||||
if res.conn != nil {
|
||||
if err := res.conn.Close(); err != nil {
|
||||
log.Debugf("failed to close losing relay connection: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
450
shared/relay/client/fallback_opener_test.go
Normal file
450
shared/relay/client/fallback_opener_test.go
Normal file
@@ -0,0 +1,450 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// The FallbackOpener race is driven by a single goroutine (Run's select loop)
|
||||
// with worker goroutines that communicate only through the buffered results
|
||||
// channel and the two cancel contexts. These tests exercise that state machine
|
||||
// in isolation via an injected openFn, so no relay server or network is needed.
|
||||
// Timing is scaled down through the fallbackDelay/totalTimeout fields.
|
||||
|
||||
// raceFakeConn tracks whether Close was called. Only Close is exercised by the
|
||||
// race logic (drainLoser closes losers; Run returns the winner untouched).
|
||||
type raceFakeConn struct {
|
||||
net.Conn
|
||||
label string
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
func (c *raceFakeConn) Close() error {
|
||||
c.closed.Store(true)
|
||||
return nil
|
||||
}
|
||||
|
||||
// raceAttemptScript describes how a single scripted attempt behaves.
|
||||
type raceAttemptScript struct {
|
||||
delay time.Duration
|
||||
conn *raceFakeConn // non-nil => the attempt succeeds and returns this conn
|
||||
err error // returned when conn is nil
|
||||
// ignoreCtx makes the attempt complete after delay even if its context is
|
||||
// cancelled. It models an OpenConn that produced a real connection right as
|
||||
// the race cancelled it - exactly the case drainLoser must clean up.
|
||||
ignoreCtx bool
|
||||
}
|
||||
|
||||
// fakeOpener replaces FallbackOpener.open. Scripts are keyed by the foreign
|
||||
// flag, so which script is "preferred" depends on the preferForeign argument
|
||||
// passed to Run.
|
||||
type fakeOpener struct {
|
||||
mu sync.Mutex
|
||||
scripts map[bool]raceAttemptScript
|
||||
calls []bool // foreign flag of each open() invocation, in order
|
||||
}
|
||||
|
||||
func (f *fakeOpener) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
|
||||
f.mu.Lock()
|
||||
f.calls = append(f.calls, foreign)
|
||||
s, ok := f.scripts[foreign]
|
||||
f.mu.Unlock()
|
||||
if !ok {
|
||||
return raceAttempt{err: fmt.Errorf("no script for foreign=%v", foreign)}
|
||||
}
|
||||
|
||||
if s.delay > 0 {
|
||||
timer := time.NewTimer(s.delay)
|
||||
defer timer.Stop()
|
||||
if s.ignoreCtx {
|
||||
<-timer.C
|
||||
} else {
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-ctx.Done():
|
||||
return raceAttempt{err: ctx.Err()}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if s.conn != nil {
|
||||
return raceAttempt{conn: s.conn}
|
||||
}
|
||||
return raceAttempt{err: s.err}
|
||||
}
|
||||
|
||||
func (f *fakeOpener) callCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.calls)
|
||||
}
|
||||
|
||||
func (f *fakeOpener) firstCallForeign(t *testing.T) bool {
|
||||
t.Helper()
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
require.NotEmpty(t, f.calls, "expected at least one open attempt")
|
||||
return f.calls[0]
|
||||
}
|
||||
|
||||
func newTestOpener(f *fakeOpener, fallbackDelay, totalTimeout time.Duration) *FallbackOpener {
|
||||
o := NewFallbackOpener(nil, nil)
|
||||
o.openFn = f.open
|
||||
o.fallbackDelay = fallbackDelay
|
||||
o.totalTimeout = totalTimeout
|
||||
return o
|
||||
}
|
||||
|
||||
const (
|
||||
// controller prefers the home relay, i.e. preferForeign == false.
|
||||
preferHome = false
|
||||
preferForeign = true
|
||||
)
|
||||
|
||||
var errAttempt = errors.New("attempt failed")
|
||||
|
||||
// The preferred attempt wins before the fallback timer fires, so the other
|
||||
// attempt is never started.
|
||||
func TestFallbackOpener_PreferredWinsImmediately(t *testing.T) {
|
||||
homeConn := &raceFakeConn{label: "home"}
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {conn: homeConn}, // preferred (home): instant success
|
||||
true: {delay: 5 * time.Second, conn: foreignConn}, // would never finish in time
|
||||
}}
|
||||
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, homeConn, conn)
|
||||
assert.Equal(t, 1, f.callCount(), "other attempt must not start when preferred wins first")
|
||||
assert.False(t, f.firstCallForeign(t), "home must be tried first when preferring home")
|
||||
assert.False(t, foreignConn.closed.Load())
|
||||
}
|
||||
|
||||
// preferForeign flips which relay is tried first.
|
||||
func TestFallbackOpener_PreferForeignRoutesForeignFirst(t *testing.T) {
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
true: {conn: foreignConn}, // preferred (foreign): instant success
|
||||
false: {delay: 5 * time.Second, conn: &raceFakeConn{}},
|
||||
}}
|
||||
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferForeign)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, foreignConn, conn)
|
||||
assert.Equal(t, 1, f.callCount())
|
||||
assert.True(t, f.firstCallForeign(t), "foreign must be tried first when preferring foreign")
|
||||
}
|
||||
|
||||
// ErrConnAlreadyExists counts as success: Run returns it and does not start the
|
||||
// other attempt.
|
||||
func TestFallbackOpener_ErrConnAlreadyExistsIsSuccess(t *testing.T) {
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {err: ErrConnAlreadyExists},
|
||||
true: {delay: 5 * time.Second, conn: &raceFakeConn{}},
|
||||
}}
|
||||
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.ErrorIs(t, err, ErrConnAlreadyExists)
|
||||
assert.Nil(t, conn)
|
||||
assert.Equal(t, 1, f.callCount(), "other attempt must not start on ErrConnAlreadyExists")
|
||||
}
|
||||
|
||||
// A preferred failure starts the other attempt immediately, without waiting for
|
||||
// the fallback timer.
|
||||
func TestFallbackOpener_PreferredFailsStartsOtherBeforeTimer(t *testing.T) {
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {err: errAttempt}, // preferred fails instantly
|
||||
true: {delay: 5 * time.Millisecond, conn: foreignConn},
|
||||
}}
|
||||
fallbackDelay := 500 * time.Millisecond
|
||||
o := newTestOpener(f, fallbackDelay, 2*time.Second)
|
||||
|
||||
start := time.Now()
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, foreignConn, conn)
|
||||
assert.Equal(t, 2, f.callCount())
|
||||
assert.Less(t, elapsed, fallbackDelay/2, "fallback must not wait for the timer after a preferred failure")
|
||||
}
|
||||
|
||||
// When the preferred attempt is slow, the fallback timer starts the other
|
||||
// attempt and its success wins.
|
||||
func TestFallbackOpener_TimerStartsOtherWhenPreferredSlow(t *testing.T) {
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {delay: 5 * time.Second}, // preferred hangs until cancelled
|
||||
true: {delay: 5 * time.Millisecond, conn: foreignConn},
|
||||
}}
|
||||
fallbackDelay := 40 * time.Millisecond
|
||||
o := newTestOpener(f, fallbackDelay, 2*time.Second)
|
||||
|
||||
start := time.Now()
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, foreignConn, conn)
|
||||
assert.Equal(t, 2, f.callCount())
|
||||
assert.GreaterOrEqual(t, elapsed, fallbackDelay, "other must not start before the fallback timer fires")
|
||||
}
|
||||
|
||||
// Both attempts fail: Run returns the last error and tries both relays.
|
||||
func TestFallbackOpener_BothFail(t *testing.T) {
|
||||
errOther := errors.New("other failed")
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {err: errAttempt},
|
||||
true: {err: errOther},
|
||||
}}
|
||||
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, conn)
|
||||
assert.ErrorIs(t, err, errOther, "the most recent error should be surfaced")
|
||||
assert.Equal(t, 2, f.callCount())
|
||||
}
|
||||
|
||||
// When both attempts succeed, drainLoser must close the losing connection so it
|
||||
// is not leaked. Here the preferred attempt wins and the foreign loser - which
|
||||
// produced a real conn despite being cancelled - is closed.
|
||||
func TestFallbackOpener_DoubleSuccessClosesLoser(t *testing.T) {
|
||||
homeConn := &raceFakeConn{label: "home"}
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {delay: 30 * time.Millisecond, conn: homeConn}, // preferred wins
|
||||
true: {delay: 80 * time.Millisecond, conn: foreignConn, ignoreCtx: true}, // loser yields a conn after cancel
|
||||
}}
|
||||
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, homeConn, conn)
|
||||
assert.Equal(t, 2, f.callCount())
|
||||
assert.False(t, homeConn.closed.Load(), "the winning connection must not be closed")
|
||||
require.Eventually(t, foreignConn.closed.Load, time.Second, 5*time.Millisecond,
|
||||
"the losing connection must be closed by drainLoser")
|
||||
}
|
||||
|
||||
// Winner selection is purely by result arrival order, not by preference: when
|
||||
// the non-preferred attempt returns first it wins even though home was
|
||||
// preferred. This is the mechanism behind the split-relay concern - two peers
|
||||
// racing independently have no shared tie-break, so under adversarial timing
|
||||
// they can settle on different relays. Documented here as current behavior.
|
||||
func TestFallbackOpener_FasterOtherWinsDespitePreference(t *testing.T) {
|
||||
homeConn := &raceFakeConn{label: "home"}
|
||||
foreignConn := &raceFakeConn{label: "foreign"}
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {delay: 60 * time.Millisecond, conn: homeConn, ignoreCtx: true}, // preferred but slower
|
||||
true: {delay: 5 * time.Millisecond, conn: foreignConn}, // other is faster
|
||||
}}
|
||||
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Same(t, foreignConn, conn, "the first successful attempt wins regardless of preference")
|
||||
require.Eventually(t, homeConn.closed.Load, time.Second, 5*time.Millisecond,
|
||||
"the slower preferred attempt becomes the loser and is closed")
|
||||
}
|
||||
|
||||
// The whole race is bounded by totalTimeout. With no attempt succeeding or
|
||||
// failing, Run returns the deadline error.
|
||||
func TestFallbackOpener_TotalTimeout(t *testing.T) {
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {delay: 5 * time.Second},
|
||||
true: {delay: 5 * time.Second},
|
||||
}}
|
||||
o := newTestOpener(f, 20*time.Millisecond, 80*time.Millisecond)
|
||||
|
||||
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, conn)
|
||||
assert.ErrorIs(t, err, context.DeadlineExceeded)
|
||||
}
|
||||
|
||||
// Cancelling the caller's context aborts the race promptly with the cancel
|
||||
// error, even before the fallback timer would fire.
|
||||
func TestFallbackOpener_ParentContextCanceled(t *testing.T) {
|
||||
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
|
||||
false: {delay: 5 * time.Second},
|
||||
true: {delay: 5 * time.Second},
|
||||
}}
|
||||
// fallbackDelay large so the timer never fires; only the parent cancel ends the race.
|
||||
o := newTestOpener(f, 5*time.Second, 5*time.Second)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
cancel()
|
||||
}()
|
||||
|
||||
start := time.Now()
|
||||
conn, err := o.Run(ctx, "peer", RelayServer{Addr: "srv"}, preferHome)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, conn)
|
||||
assert.ErrorIs(t, err, context.Canceled)
|
||||
assert.Less(t, time.Since(start), time.Second, "must return shortly after the parent context is cancelled")
|
||||
assert.Equal(t, 1, f.callCount(), "the other attempt must not start")
|
||||
}
|
||||
|
||||
// rendezvous models the relay-level requirement that a relayed connection is
|
||||
// established only once BOTH peers subscribe to the same relay server. arrive
|
||||
// records a peer's presence on a relay and returns a channel that closes when
|
||||
// the second peer arrives, so an attempt can only complete after a real
|
||||
// rendezvous - the same coupling the production code depends on.
|
||||
type rendezvous struct {
|
||||
mu sync.Mutex
|
||||
arrivals map[string]int
|
||||
gates map[string]chan struct{}
|
||||
}
|
||||
|
||||
func newRendezvous() *rendezvous {
|
||||
return &rendezvous{arrivals: map[string]int{}, gates: map[string]chan struct{}{}}
|
||||
}
|
||||
|
||||
func (r *rendezvous) arrive(relay string) <-chan struct{} {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
g, ok := r.gates[relay]
|
||||
if !ok {
|
||||
g = make(chan struct{})
|
||||
r.gates[relay] = g
|
||||
}
|
||||
r.arrivals[relay]++
|
||||
if r.arrivals[relay] == 2 {
|
||||
close(g)
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// splitPeer is one peer's view of the two relays. relayFor maps the foreign
|
||||
// flag to a relay name; postDelay is how long after the rendezvous that peer's
|
||||
// OpenConn takes to return (its per-relay subscribe latency). Different values
|
||||
// per peer model the asymmetric timing that triggers finding #1.
|
||||
type splitPeer struct {
|
||||
rv *rendezvous
|
||||
relayFor map[bool]string
|
||||
postDelay map[string]time.Duration
|
||||
}
|
||||
|
||||
func (p *splitPeer) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
|
||||
relay := p.relayFor[foreign]
|
||||
|
||||
select {
|
||||
case <-p.rv.arrive(relay):
|
||||
case <-ctx.Done():
|
||||
return raceAttempt{err: ctx.Err()}
|
||||
}
|
||||
|
||||
timer := time.NewTimer(p.postDelay[relay])
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-ctx.Done():
|
||||
return raceAttempt{err: ctx.Err()}
|
||||
}
|
||||
return raceAttempt{conn: &raceFakeConn{label: relay}}
|
||||
}
|
||||
|
||||
// TestFallbackOpener_SplitRelaySelection reproduces finding #1: the two peers
|
||||
// run FallbackOpener.Run independently with no shared tie-break, so the winner
|
||||
// is chosen purely by local result-arrival order. Under an adversarial - but
|
||||
// self-consistent - timing profile they settle on DIFFERENT relays.
|
||||
//
|
||||
// Both peers prefer relayA (the controller's home). The split needs each peer's
|
||||
// preferred relayA to be slow enough that both start their fallback (so both
|
||||
// relays actually rendezvous), and then each peer's fast path to be a different
|
||||
// relay:
|
||||
// - peerA: relayA slow (abandoned), relayB fast -> peerA wins relayB
|
||||
// - peerB: relayA fast (wins), relayB slow -> peerB wins relayA
|
||||
//
|
||||
// Each winner then cancels its attempt on the relay the OTHER peer actually
|
||||
// kept, leaving two half-open relayed connections that were both reported as
|
||||
// successful. When a deterministic cross-peer tie-break is added to fix this,
|
||||
// invert the assertion below to require convergence.
|
||||
func TestFallbackOpener_SplitRelaySelection(t *testing.T) {
|
||||
const (
|
||||
relayA = "relayA" // controller's home relay; both peers prefer it
|
||||
relayB = "relayB" // non-controller's home relay
|
||||
)
|
||||
rv := newRendezvous()
|
||||
|
||||
peerA := &splitPeer{
|
||||
rv: rv,
|
||||
relayFor: map[bool]string{false: relayA, true: relayB}, // home=relayA
|
||||
postDelay: map[string]time.Duration{
|
||||
relayA: 500 * time.Millisecond, // preferred but slow -> abandoned
|
||||
relayB: 10 * time.Millisecond, // fallback is fast -> peerA wins relayB
|
||||
},
|
||||
}
|
||||
peerB := &splitPeer{
|
||||
rv: rv,
|
||||
relayFor: map[bool]string{false: relayB, true: relayA}, // home=relayB
|
||||
postDelay: map[string]time.Duration{
|
||||
relayA: 80 * time.Millisecond, // preferred, wins - but only after starting fallback
|
||||
relayB: 500 * time.Millisecond, // fallback (home) is slow -> abandoned
|
||||
},
|
||||
}
|
||||
|
||||
fallbackDelay := 30 * time.Millisecond
|
||||
newPeerOpener := func(p *splitPeer) *FallbackOpener {
|
||||
o := NewFallbackOpener(nil, nil)
|
||||
o.openFn = p.open
|
||||
o.fallbackDelay = fallbackDelay
|
||||
o.totalTimeout = 5 * time.Second
|
||||
return o
|
||||
}
|
||||
oA := newPeerOpener(peerA)
|
||||
oB := newPeerOpener(peerB)
|
||||
|
||||
type result struct {
|
||||
conn net.Conn
|
||||
err error
|
||||
}
|
||||
var ra, rb result
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
ra.conn, ra.err = oA.Run(context.Background(), "peerB", RelayServer{Addr: relayB}, preferHome)
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
rb.conn, rb.err = oB.Run(context.Background(), "peerA", RelayServer{Addr: relayA}, preferForeign)
|
||||
}()
|
||||
wg.Wait()
|
||||
|
||||
require.NoError(t, ra.err)
|
||||
require.NoError(t, rb.err)
|
||||
aRelay := ra.conn.(*raceFakeConn).label
|
||||
bRelay := rb.conn.(*raceFakeConn).label
|
||||
t.Logf("peerA settled on %s, peerB settled on %s", aRelay, bRelay)
|
||||
|
||||
assert.Equal(t, aRelay, bRelay,
|
||||
"peers selected different relays with no cross-peer tie-break")
|
||||
assert.Equal(t, relayB, aRelay, "peerA abandoned its slow preferred relay and won the fallback")
|
||||
assert.Equal(t, relayA, bRelay, "peerB won its preferred relay after starting the fallback")
|
||||
}
|
||||
155
shared/relay/client/foreign_relays_store.go
Normal file
155
shared/relay/client/foreign_relays_store.go
Normal file
@@ -0,0 +1,155 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/sync/singleflight"
|
||||
|
||||
relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
|
||||
)
|
||||
|
||||
type foreignRelay struct {
|
||||
client *Client
|
||||
created time.Time
|
||||
inUse int
|
||||
}
|
||||
|
||||
type ForeignRelaysStore struct {
|
||||
mu sync.RWMutex
|
||||
clients map[string]*foreignRelay
|
||||
|
||||
group singleflight.Group
|
||||
|
||||
ctx context.Context
|
||||
tokenStore *relayAuth.TokenStore
|
||||
peerID string
|
||||
mtu uint16
|
||||
transportFallback *transportFallback
|
||||
onDisconnect func(string)
|
||||
keepUnusedServerTime time.Duration
|
||||
}
|
||||
|
||||
func NewForeignRelaysStore(ctx context.Context, tokenStore *relayAuth.TokenStore, peerID string, mtu uint16, transportFallback *transportFallback, onDisconnect func(string), keepUnusedServerTime time.Duration) *ForeignRelaysStore {
|
||||
return &ForeignRelaysStore{
|
||||
clients: make(map[string]*foreignRelay),
|
||||
ctx: ctx,
|
||||
tokenStore: tokenStore,
|
||||
peerID: peerID,
|
||||
mtu: mtu,
|
||||
transportFallback: transportFallback,
|
||||
onDisconnect: onDisconnect,
|
||||
keepUnusedServerTime: keepUnusedServerTime,
|
||||
}
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) OpenConn(ctx context.Context, peerKey string, remoteRelayServer RelayServer) (net.Conn, error) {
|
||||
fr, err := f.acquire(remoteRelayServer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.release(fr)
|
||||
|
||||
return fr.client.OpenConn(ctx, peerKey)
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) acquire(remoteRelayServer RelayServer) (*foreignRelay, error) {
|
||||
f.mu.Lock()
|
||||
if fr, ok := f.clients[remoteRelayServer.Addr]; ok {
|
||||
fr.inUse++
|
||||
f.mu.Unlock()
|
||||
return fr, nil
|
||||
}
|
||||
f.mu.Unlock()
|
||||
|
||||
v, err, _ := f.group.Do(remoteRelayServer.Addr, func() (any, error) {
|
||||
f.mu.RLock()
|
||||
fr, ok := f.clients[remoteRelayServer.Addr]
|
||||
f.mu.RUnlock()
|
||||
if ok {
|
||||
return fr, nil
|
||||
}
|
||||
|
||||
relayClient := NewClientWithServerIP(remoteRelayServer.Addr, remoteRelayServer.IP, f.tokenStore, f.peerID, f.mtu)
|
||||
relayClient.SetTransportFallback(f.transportFallback)
|
||||
if err := relayClient.Connect(f.ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
relayClient.SetOnDisconnectListener(f.onDisconnect)
|
||||
|
||||
f.mu.Lock()
|
||||
fr = &foreignRelay{client: relayClient, created: time.Now()}
|
||||
f.clients[remoteRelayServer.Addr] = fr
|
||||
f.mu.Unlock()
|
||||
return fr, nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
fr := v.(*foreignRelay)
|
||||
f.mu.Lock()
|
||||
if cur, ok := f.clients[remoteRelayServer.Addr]; !ok || cur != fr {
|
||||
f.mu.Unlock()
|
||||
return f.acquire(remoteRelayServer)
|
||||
}
|
||||
fr.inUse++
|
||||
f.mu.Unlock()
|
||||
return fr, nil
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) release(fr *foreignRelay) {
|
||||
f.mu.Lock()
|
||||
fr.inUse--
|
||||
f.mu.Unlock()
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) evict(serverAddress string) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if _, ok := f.clients[serverAddress]; ok {
|
||||
delete(f.clients, serverAddress)
|
||||
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) cleanupUnused() {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
|
||||
for addr, fr := range f.clients {
|
||||
if time.Since(fr.created) <= f.keepUnusedServerTime {
|
||||
continue
|
||||
}
|
||||
if fr.inUse > 0 {
|
||||
continue
|
||||
}
|
||||
if fr.client.HasConns() {
|
||||
continue
|
||||
}
|
||||
fr.client.SetOnDisconnectListener(nil)
|
||||
go func() {
|
||||
_ = fr.client.Close()
|
||||
}()
|
||||
log.Debugf("clean up unused relay server connection: %s", addr)
|
||||
delete(f.clients, addr)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *ForeignRelaysStore) states() []RelayConnState {
|
||||
f.mu.RLock()
|
||||
clients := make([]*Client, 0, len(f.clients))
|
||||
for _, fr := range f.clients {
|
||||
clients = append(clients, fr.client)
|
||||
}
|
||||
f.mu.RUnlock()
|
||||
|
||||
states := make([]RelayConnState, 0, len(clients))
|
||||
for _, c := range clients {
|
||||
states = append(states, relayConnState(c))
|
||||
}
|
||||
return states
|
||||
}
|
||||
@@ -22,27 +22,6 @@ var (
|
||||
ErrRelayClientNotConnected = fmt.Errorf("relay client not connected")
|
||||
)
|
||||
|
||||
// RelayTrack hold the relay clients for the foreign relay servers.
|
||||
// With the mutex can ensure we can open new connection in case the relay connection has been established with
|
||||
// the relay server.
|
||||
type RelayTrack struct {
|
||||
sync.RWMutex
|
||||
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{}),
|
||||
}
|
||||
}
|
||||
|
||||
type OnServerCloseListener func()
|
||||
|
||||
// ManagerOption configures a Manager at construction time.
|
||||
@@ -59,6 +38,11 @@ type RelayConnState struct {
|
||||
Err error
|
||||
}
|
||||
|
||||
type RelayServer struct {
|
||||
Addr string
|
||||
IP netip.Addr
|
||||
}
|
||||
|
||||
// WithMaxBackoffInterval caps the exponential backoff between reconnect
|
||||
// attempts to the home relay. A non-positive value keeps the default.
|
||||
func WithMaxBackoffInterval(d time.Duration) ManagerOption {
|
||||
@@ -83,8 +67,7 @@ type Manager struct {
|
||||
relayClientMu sync.RWMutex
|
||||
reconnectGuard *Guard
|
||||
|
||||
relayClients map[string]*RelayTrack
|
||||
relayClientsMutex sync.RWMutex
|
||||
foreign *ForeignRelaysStore
|
||||
|
||||
onDisconnectedListeners map[string]*list.List
|
||||
onReconnectedListenerFn func()
|
||||
@@ -120,7 +103,6 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
|
||||
ConnectionTimeout: defaultConnectionTimeout,
|
||||
TransportFallback: tf,
|
||||
},
|
||||
relayClients: make(map[string]*RelayTrack),
|
||||
onDisconnectedListeners: make(map[string]*list.List),
|
||||
cleanupInterval: relayCleanupInterval,
|
||||
keepUnusedServerTime: keepUnusedServerTime,
|
||||
@@ -128,6 +110,7 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
|
||||
for _, opt := range opts {
|
||||
opt(m)
|
||||
}
|
||||
m.foreign = NewForeignRelaysStore(ctx, tokenStore, peerID, mtu, tf, m.onServerDisconnected, m.keepUnusedServerTime)
|
||||
m.serverPicker.ServerURLs.Store(serverURLs)
|
||||
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval)
|
||||
return m
|
||||
@@ -159,40 +142,26 @@ func (m *Manager) Serve() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// OpenConn opens a connection to the given peer key. If the peer is on the same relay server, the connection will be
|
||||
// established via the relay server. If the peer is on a different relay server, the manager will establish a new
|
||||
// connection to the relay server. It returns back with a net.Conn what represent the remote peer connection.
|
||||
//
|
||||
// serverIP, when valid and serverAddress is foreign, is used as a dial target if the FQDN-based dial fails.
|
||||
// Ignored for the local home-server path. TLS verification still uses the FQDN via SNI.
|
||||
func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
|
||||
func (m *Manager) OpenConn(ctx context.Context, remoteRelayServer RelayServer, peerKey string, preferForeign bool) (net.Conn, error) {
|
||||
m.relayClientMu.RLock()
|
||||
defer m.relayClientMu.RUnlock()
|
||||
relayClient := m.relayClient
|
||||
m.relayClientMu.RUnlock()
|
||||
|
||||
if m.relayClient == nil {
|
||||
if relayClient == nil {
|
||||
return nil, ErrRelayClientNotConnected
|
||||
}
|
||||
|
||||
foreign, err := m.isForeignServer(serverAddress)
|
||||
foreign, err := m.isForeignServer(relayClient, remoteRelayServer.Addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var (
|
||||
netConn net.Conn
|
||||
)
|
||||
if !foreign {
|
||||
log.Debugf("open peer connection via permanent server: %s", peerKey)
|
||||
netConn, err = m.relayClient.OpenConn(ctx, peerKey)
|
||||
} else {
|
||||
log.Debugf("open peer connection via foreign server: %s", serverAddress)
|
||||
netConn, err = m.openConnVia(ctx, serverAddress, peerKey, serverIP)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
|
||||
return netConn, err
|
||||
opener := NewFallbackOpener(relayClient, m.foreign)
|
||||
return opener.Run(ctx, peerKey, remoteRelayServer, preferForeign)
|
||||
}
|
||||
|
||||
// Ready returns true if the home Relay client is connected to the relay server.
|
||||
@@ -223,7 +192,7 @@ func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServ
|
||||
return ErrRelayClientNotConnected
|
||||
}
|
||||
|
||||
foreign, err := m.isForeignServer(serverAddress)
|
||||
foreign, err := m.isForeignServer(m.relayClient, serverAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -287,26 +256,7 @@ func (m *Manager) RelayStates() []RelayConnState {
|
||||
states = append(states, st)
|
||||
}
|
||||
|
||||
// Snapshot the tracks, then query each outside the map lock: a track can be
|
||||
// held by an in-progress Connect, and blocking on it must not stall other
|
||||
// relay operations.
|
||||
m.relayClientsMutex.RLock()
|
||||
tracks := make([]*RelayTrack, 0, len(m.relayClients))
|
||||
for _, rt := range m.relayClients {
|
||||
tracks = append(tracks, rt)
|
||||
}
|
||||
m.relayClientsMutex.RUnlock()
|
||||
|
||||
// Only connected foreign relays carry state; a failed connect is evicted
|
||||
// immediately (openConnVia), so there is no error state to surface.
|
||||
for _, rt := range tracks {
|
||||
rt.RLock()
|
||||
rc := rt.relayClient
|
||||
rt.RUnlock()
|
||||
if rc != nil {
|
||||
states = append(states, relayConnState(rc))
|
||||
}
|
||||
}
|
||||
states = append(states, m.foreign.states()...)
|
||||
|
||||
return states
|
||||
}
|
||||
@@ -327,76 +277,6 @@ func (m *Manager) UpdateToken(token *relayAuth.Token) error {
|
||||
return m.tokenStore.UpdateToken(token)
|
||||
}
|
||||
|
||||
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
|
||||
// check if already has a connection to the desired relay server
|
||||
m.relayClientsMutex.RLock()
|
||||
rt, ok := m.relayClients[serverAddress]
|
||||
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 {
|
||||
m.relayClientsMutex.Unlock()
|
||||
return m.openConnOnTrack(ctx, rt, peerKey)
|
||||
}
|
||||
|
||||
// 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()
|
||||
m.relayClients[serverAddress] = rt
|
||||
m.relayClientsMutex.Unlock()
|
||||
|
||||
relayClient := NewClientWithServerIP(serverAddress, serverIP, m.tokenStore, m.peerID, m.mtu)
|
||||
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()
|
||||
return nil, err
|
||||
}
|
||||
// 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)
|
||||
|
||||
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()
|
||||
}
|
||||
|
||||
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() {
|
||||
m.listenerLock.Lock()
|
||||
defer m.listenerLock.Unlock()
|
||||
@@ -422,21 +302,12 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
|
||||
m.relayClientMu.Unlock()
|
||||
|
||||
if !isHome {
|
||||
m.evictForeignRelay(serverAddress)
|
||||
m.foreign.evict(serverAddress)
|
||||
}
|
||||
|
||||
m.notifyOnDisconnectListeners(serverAddress)
|
||||
}
|
||||
|
||||
func (m *Manager) evictForeignRelay(serverAddress string) {
|
||||
m.relayClientsMutex.Lock()
|
||||
defer m.relayClientsMutex.Unlock()
|
||||
if _, ok := m.relayClients[serverAddress]; ok {
|
||||
delete(m.relayClients, serverAddress)
|
||||
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) listenGuardEvent(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
@@ -459,8 +330,8 @@ func (m *Manager) storeClient(client *Client) {
|
||||
m.relayClient.SetOnDisconnectListener(m.onServerDisconnected)
|
||||
}
|
||||
|
||||
func (m *Manager) isForeignServer(address string) (bool, error) {
|
||||
rAddr, err := m.relayClient.ServerInstanceURL()
|
||||
func (m *Manager) isForeignServer(relayClient *Client, address string) (bool, error) {
|
||||
rAddr, err := relayClient.ServerInstanceURL()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("relay client not connected")
|
||||
}
|
||||
@@ -475,50 +346,11 @@ func (m *Manager) startCleanupLoop() {
|
||||
case <-m.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
m.cleanUpUnusedRelays()
|
||||
m.foreign.cleanupUnused()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) cleanUpUnusedRelays() {
|
||||
m.relayClientsMutex.Lock()
|
||||
defer m.relayClientsMutex.Unlock()
|
||||
|
||||
for addr, rt := range m.relayClients {
|
||||
rt.Lock()
|
||||
// if the connection failed to the server the relay client will be nil
|
||||
// but the instance will be kept in the relayClients until the next locking
|
||||
if rt.err != nil {
|
||||
rt.Unlock()
|
||||
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
|
||||
}
|
||||
|
||||
if rt.relayClient.HasConns() {
|
||||
rt.Unlock()
|
||||
continue
|
||||
}
|
||||
rt.relayClient.SetOnDisconnectListener(nil)
|
||||
go func() {
|
||||
_ = rt.relayClient.Close()
|
||||
}()
|
||||
log.Debugf("clean up unused relay server connection: %s", addr)
|
||||
delete(m.relayClients, addr)
|
||||
rt.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) addListener(serverAddress string, onClosedListener OnServerCloseListener) {
|
||||
m.listenerLock.Lock()
|
||||
defer m.listenerLock.Unlock()
|
||||
|
||||
@@ -2,17 +2,14 @@ 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.
|
||||
// relay dial and asserts the foreign store cleanup does not stall behind it.
|
||||
func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
serverAddr := stallingRelayListener(t)
|
||||
serverAddr, accepted := stallingRelayListener(t)
|
||||
|
||||
mCtx, mCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(mCancel)
|
||||
@@ -22,39 +19,32 @@ func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
dialDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(dialDone)
|
||||
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
|
||||
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
|
||||
}()
|
||||
|
||||
// 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")
|
||||
select {
|
||||
case <-accepted:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("relay dial did not reach the listener")
|
||||
}
|
||||
|
||||
cleanupDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(cleanupDone)
|
||||
m.cleanUpUnusedRelays()
|
||||
m.foreign.cleanupUnused()
|
||||
}()
|
||||
|
||||
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")
|
||||
t.Fatal("cleanupUnused blocked on an in-progress relay dial")
|
||||
}
|
||||
|
||||
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")
|
||||
t.Fatal("foreign OpenConn did not return after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package client
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -13,13 +12,16 @@ import (
|
||||
|
||||
// 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 {
|
||||
// cancelled. accepted is signalled once per incoming connection so a caller can
|
||||
// wait until a dial has actually reached the listener. It returns the
|
||||
// "rel://host:port" URL to dial.
|
||||
func stallingRelayListener(t *testing.T) (string, <-chan struct{}) {
|
||||
t.Helper()
|
||||
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
|
||||
accepted := make(chan struct{}, 1)
|
||||
var mu sync.Mutex
|
||||
var conns []net.Conn
|
||||
go func() {
|
||||
@@ -31,6 +33,10 @@ func stallingRelayListener(t *testing.T) string {
|
||||
mu.Lock()
|
||||
conns = append(conns, c)
|
||||
mu.Unlock()
|
||||
select {
|
||||
case accepted <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
@@ -42,14 +48,14 @@ func stallingRelayListener(t *testing.T) string {
|
||||
mu.Unlock()
|
||||
})
|
||||
|
||||
return "rel://" + ln.Addr().String()
|
||||
return "rel://" + ln.Addr().String(), accepted
|
||||
}
|
||||
|
||||
// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for
|
||||
// RelayStates() called by a "status -d command" hanging behind an in-progress
|
||||
// relay dial.
|
||||
// foreign relay dial.
|
||||
func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
serverAddr := stallingRelayListener(t)
|
||||
serverAddr, accepted := stallingRelayListener(t)
|
||||
|
||||
mCtx, mCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(mCancel)
|
||||
@@ -59,15 +65,14 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
dialDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(dialDone)
|
||||
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
|
||||
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
|
||||
}()
|
||||
|
||||
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")
|
||||
select {
|
||||
case <-accepted:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("relay dial did not reach the listener")
|
||||
}
|
||||
|
||||
done := make(chan []RelayConnState, 1)
|
||||
go func() {
|
||||
@@ -86,6 +91,6 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
select {
|
||||
case <-dialDone:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("openConnVia did not return after context cancellation")
|
||||
t.Fatal("foreign OpenConn did not return after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package client
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -85,7 +84,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
|
||||
t.Run("no server IP, dial fails", func(t *testing.T) {
|
||||
dialCtx, dialCancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer dialCancel()
|
||||
_, err := mgrAlice.OpenConn(dialCtx, brokenFQDN, "bob", netip.Addr{})
|
||||
_, err := mgrAlice.OpenConn(dialCtx, RelayServer{Addr: brokenFQDN}, "bob", true)
|
||||
if err == nil {
|
||||
t.Fatalf("expected OpenConn to fail without server IP, got success")
|
||||
}
|
||||
@@ -95,7 +94,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
|
||||
// Bob waits for Alice's incoming peer connection on his side.
|
||||
bobSideCh := make(chan error, 1)
|
||||
go func() {
|
||||
conn, err := mgrBob.OpenConn(ctx, bobRealAddr, "alice", netip.Addr{})
|
||||
conn, err := mgrBob.OpenConn(ctx, RelayServer{Addr: bobRealAddr}, "alice", false)
|
||||
if err != nil {
|
||||
bobSideCh <- err
|
||||
return
|
||||
@@ -113,7 +112,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
|
||||
bobSideCh <- nil
|
||||
}()
|
||||
|
||||
aliceConn, err := mgrAlice.OpenConn(ctx, brokenFQDN, "bob", bobAdvertisedIP)
|
||||
aliceConn, err := mgrAlice.OpenConn(ctx, RelayServer{Addr: brokenFQDN, IP: bobAdvertisedIP}, "bob", true)
|
||||
if err != nil {
|
||||
t.Fatalf("alice OpenConn with server IP: %s", err)
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package client
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -106,11 +105,11 @@ func TestForeignConn(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get relay address: %s", err)
|
||||
}
|
||||
connAliceToBob, err := clientAlice.OpenConn(ctx, bobsSrvAddr, "bob", netip.Addr{})
|
||||
connAliceToBob, err := clientAlice.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "bob", true)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to bind channel: %s", err)
|
||||
}
|
||||
connBobToAlice, err := clientBob.OpenConn(ctx, bobsSrvAddr, "alice", netip.Addr{})
|
||||
connBobToAlice, err := clientBob.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "alice", false)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to bind channel: %s", err)
|
||||
}
|
||||
@@ -210,7 +209,7 @@ func TestForeginConnClose(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("failed to serve manager: %s", err)
|
||||
}
|
||||
conn, err := mgr.OpenConn(ctx, toURL(srvCfg2)[0], "bob", netip.Addr{})
|
||||
conn, err := mgr.OpenConn(ctx, RelayServer{Addr: toURL(srvCfg2)[0]}, "bob", true)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to bind channel: %s", err)
|
||||
}
|
||||
@@ -302,7 +301,7 @@ func TestForeignAutoClose(t *testing.T) {
|
||||
}
|
||||
|
||||
t.Log("open connection to another peer")
|
||||
if _, err = mgr.OpenConn(ctx, foreignServerURL, "anotherpeer", netip.Addr{}); err == nil {
|
||||
if _, err = mgr.OpenConn(ctx, RelayServer{Addr: foreignServerURL}, "anotherpeer", true); err == nil {
|
||||
t.Fatalf("should have failed to open connection to another peer")
|
||||
}
|
||||
|
||||
@@ -372,7 +371,7 @@ func TestAutoReconnect(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Errorf("failed to get relay address: %s", err)
|
||||
}
|
||||
conn, err := clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
|
||||
conn, err := clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
|
||||
if err != nil {
|
||||
t.Errorf("failed to bind channel: %s", err)
|
||||
}
|
||||
@@ -392,7 +391,7 @@ func TestAutoReconnect(t *testing.T) {
|
||||
}
|
||||
|
||||
log.Infof("reopent the connection")
|
||||
_, err = clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
|
||||
_, err = clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
|
||||
if err != nil {
|
||||
t.Errorf("failed to open channel: %s", err)
|
||||
}
|
||||
@@ -454,7 +453,7 @@ func TestNotifierDoubleAdd(t *testing.T) {
|
||||
t.Fatalf("failed to serve manager: %s", err)
|
||||
}
|
||||
|
||||
conn1, err := clientAlice.OpenConn(ctx, clientAlice.ServerURLs()[0], "bob", netip.Addr{})
|
||||
conn1, err := clientAlice.OpenConn(ctx, RelayServer{Addr: clientAlice.ServerURLs()[0]}, "bob", false)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to bind channel: %s", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user