mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-23 15:19:08 +02:00
Merge branch 'main' into embedded-vnc
# Conflicts: # client/ui/frontend/src/app.tsx # client/ui/frontend/src/modules/main/MainConnectionStatusSwitch.tsx # client/ui/i18n/locales/uk/common.json # go.sum
This commit is contained in:
@@ -135,6 +135,11 @@ func NewUserPendingApprovalError() error {
|
||||
return Errorf(PermissionDenied, "user is pending approval")
|
||||
}
|
||||
|
||||
// NewUserPendingApprovalByOwnerError creates a new Error with PermissionDenied type for a blocked user pending approval, naming the masked address of the owner who can approve them
|
||||
func NewUserPendingApprovalByOwnerError(ownerEmail string) error {
|
||||
return Errorf(PermissionDenied, "user is pending approval by owner %s", ownerEmail)
|
||||
}
|
||||
|
||||
// NewPeerNotRegisteredError creates a new Error with Unauthenticated type unregistered peer
|
||||
func NewPeerNotRegisteredError() error {
|
||||
return Errorf(Unauthenticated, "peer is not registered")
|
||||
|
||||
@@ -30,6 +30,12 @@ const (
|
||||
|
||||
var (
|
||||
ErrConnAlreadyExists = fmt.Errorf("connection already exists")
|
||||
// ErrServerDisconnected is the cancellation cause of a relayed Conn when the
|
||||
// client lost the connection to the relay server.
|
||||
ErrServerDisconnected = fmt.Errorf("relay server disconnected")
|
||||
// ErrPeerDisconnected is the cancellation cause of a relayed Conn when the
|
||||
// remote peer went offline.
|
||||
ErrPeerDisconnected = fmt.Errorf("remote peer disconnected")
|
||||
)
|
||||
|
||||
type internalStopFlag struct {
|
||||
@@ -74,16 +80,17 @@ type connContainer struct {
|
||||
msgChanLock sync.Mutex
|
||||
closed bool // flag to check if channel is closed
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
cancel context.CancelCauseFunc
|
||||
}
|
||||
|
||||
func newConnContainer(log *log.Entry, c *Client, peerID messages.PeerID, instanceURL *RelayAddr) *connContainer {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
ctx, cancel := context.WithCancelCause(context.Background())
|
||||
msgChan := make(chan Msg, connChannelSize)
|
||||
cn := &Conn{
|
||||
dstID: peerID,
|
||||
messageChan: msgChan,
|
||||
instanceURL: instanceURL,
|
||||
ctx: ctx,
|
||||
}
|
||||
cc := &connContainer{
|
||||
log: log,
|
||||
@@ -106,10 +113,6 @@ func newConnContainer(log *log.Entry, c *Client, peerID messages.PeerID, instanc
|
||||
return cc
|
||||
}
|
||||
|
||||
func (cc *connContainer) netConn() net.Conn {
|
||||
return cc.conn
|
||||
}
|
||||
|
||||
func (cc *connContainer) writeMsg(msg Msg) {
|
||||
cc.msgChanLock.Lock()
|
||||
defer cc.msgChanLock.Unlock()
|
||||
@@ -128,8 +131,8 @@ func (cc *connContainer) writeMsg(msg Msg) {
|
||||
}
|
||||
}
|
||||
|
||||
func (cc *connContainer) close() {
|
||||
cc.cancel()
|
||||
func (cc *connContainer) close(cause error) {
|
||||
cc.cancel(cause)
|
||||
|
||||
cc.msgChanLock.Lock()
|
||||
defer cc.msgChanLock.Unlock()
|
||||
@@ -279,7 +282,7 @@ func (c *Client) Connect(ctx context.Context) error {
|
||||
c.stateSubscription = NewPeersStateSubscription(c.log, c.relayConn, c.closeConnsByPeerID)
|
||||
|
||||
c.log = c.log.WithField("relay", instanceURL.String())
|
||||
c.log.Infof("relay connection established")
|
||||
c.log.Infof("relay connection established, server IP: %s", connectedIP(c.relayConn))
|
||||
|
||||
c.serviceIsRunning = true
|
||||
|
||||
@@ -293,12 +296,12 @@ func (c *Client) Connect(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// OpenConn create a new net.Conn for the destination peer ID. In case if the connection is in progress
|
||||
// OpenConn create a new Conn for the destination peer ID. In case if the connection is in progress
|
||||
// to the relay server, the function will block until the connection is established or timed out. Otherwise,
|
||||
// it will return immediately.
|
||||
// It block until the server confirm the peer is online.
|
||||
// todo: what should happen if call with the same peerID with multiple times?
|
||||
func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, error) {
|
||||
func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (*Conn, error) {
|
||||
peerID := messages.HashID(dstPeerID)
|
||||
|
||||
c.mu.Lock()
|
||||
@@ -335,7 +338,7 @@ func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, erro
|
||||
delete(c.conns, peerID)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
container.close()
|
||||
container.close(err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -345,13 +348,13 @@ func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, erro
|
||||
delete(c.conns, peerID)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
container.close()
|
||||
container.close(ErrServerDisconnected)
|
||||
return nil, fmt.Errorf("relay connection is not established")
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
c.log.Infof("remote peer is available: %s", peerID)
|
||||
return container.netConn(), nil
|
||||
return container.conn, nil
|
||||
}
|
||||
|
||||
// ServerInstanceURL returns the address of the relay server. It could change after the close and reopen the connection.
|
||||
@@ -364,23 +367,6 @@ func (c *Client) ServerInstanceURL() (string, error) {
|
||||
return c.instanceURL.String(), nil
|
||||
}
|
||||
|
||||
// ConnectedIP returns the IP address of the live relay-server connection,
|
||||
// extracted from the underlying socket's RemoteAddr. Zero value if not
|
||||
// connected or if the address is not an IP literal.
|
||||
func (c *Client) ConnectedIP() netip.Addr {
|
||||
c.mu.Lock()
|
||||
conn := c.relayConn
|
||||
c.mu.Unlock()
|
||||
if conn == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
addr := conn.RemoteAddr()
|
||||
if addr == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
return extractIPLiteral(addr.String())
|
||||
}
|
||||
|
||||
// SetOnDisconnectListener sets a function that will be called when the connection to the relay server is closed.
|
||||
func (c *Client) SetOnDisconnectListener(fn func(string)) {
|
||||
c.listenerMutex.Lock()
|
||||
@@ -777,9 +763,20 @@ func (c *Client) listenForStopEvents(ctx context.Context, hc *healthcheck.Receiv
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) serverInstanceAddress() (string, netip.Addr, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
addr, err := c.ServerInstanceURL()
|
||||
if err != nil {
|
||||
return "", netip.Addr{}, err
|
||||
}
|
||||
return addr, connectedIP(c.relayConn), nil
|
||||
}
|
||||
|
||||
func (c *Client) closeAllConns() {
|
||||
for _, container := range c.conns {
|
||||
container.close()
|
||||
container.close(ErrServerDisconnected)
|
||||
}
|
||||
c.conns = make(map[messages.PeerID]*connContainer)
|
||||
|
||||
@@ -799,7 +796,7 @@ func (c *Client) closeConnsByPeerID(peerIDs []messages.PeerID) {
|
||||
}
|
||||
|
||||
container.log.Infof("remote peer has been disconnected, free up connection: %s", peerID)
|
||||
container.close()
|
||||
container.close(ErrPeerDisconnected)
|
||||
delete(c.conns, peerID)
|
||||
}
|
||||
|
||||
@@ -827,7 +824,7 @@ func (c *Client) closeConn(containerRef *connContainer, id messages.PeerID) erro
|
||||
|
||||
c.log.Infof("free up connection to peer: %s", id)
|
||||
delete(c.conns, id)
|
||||
current.close()
|
||||
current.close(net.ErrClosed)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -923,6 +920,17 @@ func (c *Client) handlePeersWentOfflineMsg(buf []byte) {
|
||||
c.stateSubscription.OnPeersWentOffline(peersID)
|
||||
}
|
||||
|
||||
func connectedIP(conn net.Conn) netip.Addr {
|
||||
if conn == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
addr := conn.RemoteAddr()
|
||||
if addr == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
return extractIPLiteral(addr.String())
|
||||
}
|
||||
|
||||
// extractIPLiteral returns the IP from address forms produced by the relay
|
||||
// dialers (URL or host:port). Zero value if the host is not an IP.
|
||||
func extractIPLiteral(s string) netip.Addr {
|
||||
|
||||
@@ -8,6 +8,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface"
|
||||
@@ -68,18 +70,17 @@ func TestClient_ServerIPRecoversFromUnresolvableFQDN(t *testing.T) {
|
||||
if !c.Ready() {
|
||||
t.Fatalf("client not ready after connect")
|
||||
}
|
||||
if got := c.ConnectedIP(); got.String() != "127.0.0.1" {
|
||||
t.Fatalf("ConnectedIP = %q, want 127.0.0.1", got)
|
||||
}
|
||||
url, ip, err := c.serverInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, srvCfg.ExposedAddress, url, "relay URL must come from the handshake")
|
||||
assert.Equal(t, netip.MustParseAddr("127.0.0.1"), ip, "relay IP must come from the connection")
|
||||
})
|
||||
}
|
||||
|
||||
// TestClient_ConnectedIPAfterFQDNDial verifies ConnectedIP returns the
|
||||
// resolved IP after a successful FQDN-based dial. The underlying socket's
|
||||
// RemoteAddr must be exposed through the dialer wrappers; if it returns
|
||||
// the dial-time URL instead, ConnectedIP returns empty and the dial
|
||||
// IP we advertise to peers is empty too.
|
||||
func TestClient_ConnectedIPAfterFQDNDial(t *testing.T) {
|
||||
// TestClient_ServerInstanceAddressAfterFQDNDial verifies the relay address
|
||||
// includes the resolved IP after an FQDN dial. The dialer wrappers must expose
|
||||
// the socket's RemoteAddr; returning the dial-time URL would lose the IP.
|
||||
func TestClient_ServerInstanceAddressAfterFQDNDial(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -111,10 +112,10 @@ func TestClient_ConnectedIPAfterFQDNDial(t *testing.T) {
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
|
||||
got := c.ConnectedIP().String()
|
||||
if got != "127.0.0.1" && got != "::1" {
|
||||
t.Fatalf("ConnectedIP after FQDN dial = %q, want 127.0.0.1 or ::1", got)
|
||||
}
|
||||
url, ip, err := c.serverInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, srvCfg.ExposedAddress, url, "relay URL must come from the handshake")
|
||||
assert.Contains(t, []string{"127.0.0.1", "::1"}, ip.String(), "relay IP must resolve to localhost")
|
||||
}
|
||||
|
||||
func TestSubstituteHost(t *testing.T) {
|
||||
@@ -214,15 +215,12 @@ func TestSubstituteHost(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_ConnectedIPEmptyWhenNotConnected(t *testing.T) {
|
||||
c := NewClient("rel://example.invalid:80", hmacTokenStore, "x", iface.DefaultMTU)
|
||||
if got := c.ConnectedIP(); got.IsValid() {
|
||||
t.Fatalf("ConnectedIP on disconnected client = %q, want zero", got)
|
||||
}
|
||||
func TestConnectedIPNilConnection(t *testing.T) {
|
||||
assert.False(t, connectedIP(nil).IsValid(), "missing connection must not provide an IP")
|
||||
}
|
||||
|
||||
// staticAddr is a net.Addr that returns a fixed string. Used to verify
|
||||
// ConnectedIP parses RemoteAddr correctly.
|
||||
// connectedIP parses RemoteAddr correctly.
|
||||
type staticAddr struct{ s string }
|
||||
|
||||
func (a staticAddr) Network() string { return "tcp" }
|
||||
@@ -235,7 +233,7 @@ type stubConn struct {
|
||||
|
||||
func (s stubConn) RemoteAddr() net.Addr { return s.remote }
|
||||
|
||||
func TestClient_ConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
func TestConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
s string
|
||||
@@ -252,15 +250,12 @@ func TestClient_ConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
c := &Client{relayConn: stubConn{remote: staticAddr{s: tt.s}}}
|
||||
got := c.ConnectedIP()
|
||||
got := connectedIP(stubConn{remote: staticAddr{s: tt.s}})
|
||||
var gotStr string
|
||||
if got.IsValid() {
|
||||
gotStr = got.String()
|
||||
}
|
||||
if gotStr != tt.want {
|
||||
t.Errorf("ConnectedIP(%q) = %q, want %q", tt.s, gotStr, tt.want)
|
||||
}
|
||||
assert.Equal(t, tt.want, gotStr, "IP extracted from RemoteAddr %q", tt.s)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
@@ -12,11 +13,20 @@ type Conn struct {
|
||||
dstID messages.PeerID
|
||||
messageChan chan Msg
|
||||
instanceURL *RelayAddr
|
||||
ctx context.Context
|
||||
writeFn func(messages.PeerID, []byte) (int, error)
|
||||
closeFn func(messages.PeerID) error
|
||||
localAddrFn func() net.Addr
|
||||
}
|
||||
|
||||
// Context returns a context that is cancelled when the connection is torn down,
|
||||
// either by Close or by the relay client losing the server connection. The
|
||||
// cancellation cause carries the reason, see ErrServerDisconnected and
|
||||
// ErrPeerDisconnected.
|
||||
func (c *Conn) Context() context.Context {
|
||||
return c.ctx
|
||||
}
|
||||
|
||||
func (c *Conn) Write(p []byte) (n int, err error) {
|
||||
return c.writeFn(c.dstID, p)
|
||||
}
|
||||
|
||||
@@ -1,12 +1,9 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"reflect"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -43,8 +40,6 @@ func NewRelayTrack() *RelayTrack {
|
||||
}
|
||||
}
|
||||
|
||||
type OnServerCloseListener func()
|
||||
|
||||
// ManagerOption configures a Manager at construction time.
|
||||
type ManagerOption func(*Manager)
|
||||
|
||||
@@ -91,7 +86,6 @@ type Manager struct {
|
||||
relayClients map[string]*RelayTrack
|
||||
relayClientsMutex sync.RWMutex
|
||||
|
||||
onDisconnectedListeners map[string]*list.List
|
||||
onReconnectedListenerFn func()
|
||||
listenerLock sync.Mutex
|
||||
|
||||
@@ -126,10 +120,9 @@ 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,
|
||||
relayClients: make(map[string]*RelayTrack),
|
||||
cleanupInterval: relayCleanupInterval,
|
||||
keepUnusedServerTime: keepUnusedServerTime,
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(m)
|
||||
@@ -168,11 +161,11 @@ func (m *Manager) Serve() error {
|
||||
|
||||
// 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.
|
||||
// connection to the relay server. It returns the relayed connection to the remote peer.
|
||||
//
|
||||
// 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, serverAddress, peerKey string, serverIP netip.Addr) (*Conn, error) {
|
||||
m.relayClientMu.RLock()
|
||||
defer m.relayClientMu.RUnlock()
|
||||
|
||||
@@ -185,9 +178,7 @@ func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, s
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var (
|
||||
netConn net.Conn
|
||||
)
|
||||
var netConn *Conn
|
||||
if !foreign {
|
||||
log.Debugf("open peer connection via permanent server: %s", peerKey)
|
||||
netConn, err = m.relayClient.OpenConn(ctx, peerKey)
|
||||
@@ -220,31 +211,6 @@ func (m *Manager) SetOnReconnectedListener(f func()) {
|
||||
m.onReconnectedListenerFn = f
|
||||
}
|
||||
|
||||
// AddCloseListener adds a listener to the given server instance address. The listener will be called if the connection
|
||||
// closed.
|
||||
func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServerCloseListener) error {
|
||||
m.relayClientMu.RLock()
|
||||
defer m.relayClientMu.RUnlock()
|
||||
|
||||
if m.relayClient == nil {
|
||||
return ErrRelayClientNotConnected
|
||||
}
|
||||
|
||||
foreign, err := m.isForeignServer(serverAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var listenerAddr string
|
||||
if foreign {
|
||||
listenerAddr = serverAddress
|
||||
} else {
|
||||
listenerAddr = m.relayClient.connectionURL
|
||||
}
|
||||
m.addListener(listenerAddr, onClosedListener)
|
||||
return nil
|
||||
}
|
||||
|
||||
// RelayInstanceAddress returns the address and resolved IP of the permanent relay server. It could change if the
|
||||
// network connection is lost. The address is sent to the target peer to choose the common relay server for the
|
||||
// communication; the IP is sent alongside so remote peers can dial directly without their own DNS lookup. Both
|
||||
@@ -256,11 +222,7 @@ func (m *Manager) RelayInstanceAddress() (string, netip.Addr, error) {
|
||||
if m.relayClient == nil {
|
||||
return "", netip.Addr{}, ErrRelayClientNotConnected
|
||||
}
|
||||
addr, err := m.relayClient.ServerInstanceURL()
|
||||
if err != nil {
|
||||
return "", netip.Addr{}, err
|
||||
}
|
||||
return addr, m.relayClient.ConnectedIP(), nil
|
||||
return m.relayClient.serverInstanceAddress()
|
||||
}
|
||||
|
||||
// ServerURLs returns the addresses of the relay servers.
|
||||
@@ -334,7 +296,7 @@ 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) {
|
||||
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (*Conn, error) {
|
||||
// check if already has a connection to the desired relay server
|
||||
m.relayClientsMutex.RLock()
|
||||
rt, ok := m.relayClients[serverAddress]
|
||||
@@ -387,7 +349,7 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
// 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) {
|
||||
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (*Conn, error) {
|
||||
select {
|
||||
case <-rt.ready:
|
||||
case <-ctx.Done():
|
||||
@@ -432,8 +394,6 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
|
||||
if !isHome {
|
||||
m.evictForeignRelay(serverAddress)
|
||||
}
|
||||
|
||||
m.notifyOnDisconnectListeners(serverAddress)
|
||||
}
|
||||
|
||||
func (m *Manager) evictForeignRelay(serverAddress string) {
|
||||
@@ -527,36 +487,6 @@ func (m *Manager) cleanUpUnusedRelays() {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) addListener(serverAddress string, onClosedListener OnServerCloseListener) {
|
||||
m.listenerLock.Lock()
|
||||
defer m.listenerLock.Unlock()
|
||||
l, ok := m.onDisconnectedListeners[serverAddress]
|
||||
if !ok {
|
||||
l = list.New()
|
||||
}
|
||||
for e := l.Front(); e != nil; e = e.Next() {
|
||||
if reflect.ValueOf(e.Value).Pointer() == reflect.ValueOf(onClosedListener).Pointer() {
|
||||
return
|
||||
}
|
||||
}
|
||||
l.PushBack(onClosedListener)
|
||||
m.onDisconnectedListeners[serverAddress] = l
|
||||
}
|
||||
|
||||
func (m *Manager) notifyOnDisconnectListeners(serverAddress string) {
|
||||
m.listenerLock.Lock()
|
||||
defer m.listenerLock.Unlock()
|
||||
|
||||
l, ok := m.onDisconnectedListeners[serverAddress]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
for e := l.Front(); e != nil; e = e.Next() {
|
||||
go e.Value.(OnServerCloseListener)()
|
||||
}
|
||||
delete(m.onDisconnectedListeners, serverAddress)
|
||||
}
|
||||
|
||||
func relayConnState(c *Client) RelayConnState {
|
||||
addr, err := c.ServerInstanceURL()
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"runtime"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestManager_RelayInstanceAddressAcrossReconnect(t *testing.T) {
|
||||
relays := []struct {
|
||||
url *RelayAddr
|
||||
conn stubConn
|
||||
ip netip.Addr
|
||||
}{
|
||||
{
|
||||
url: &RelayAddr{addr: "rels://relay-a.example:443"},
|
||||
conn: stubConn{remote: staticAddr{s: "192.0.2.1:443"}},
|
||||
ip: netip.MustParseAddr("192.0.2.1"),
|
||||
},
|
||||
{
|
||||
url: &RelayAddr{addr: "rels://relay-b.example:443"},
|
||||
conn: stubConn{remote: staticAddr{s: "192.0.2.2:443"}},
|
||||
ip: netip.MustParseAddr("192.0.2.2"),
|
||||
},
|
||||
}
|
||||
c := &Client{
|
||||
instanceURL: relays[0].url,
|
||||
relayConn: relays[0].conn,
|
||||
serviceIsRunning: true,
|
||||
}
|
||||
m := &Manager{relayClient: c}
|
||||
started := make(chan struct{})
|
||||
stop := make(chan struct{})
|
||||
done := make(chan struct{})
|
||||
t.Cleanup(func() {
|
||||
close(stop)
|
||||
<-done
|
||||
})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 0; ; i++ {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
// Publish successive connection states using the lifecycle locks.
|
||||
// Yield before publication so a getter using only muInstanceURL
|
||||
// can read the old URL while waiting for the new connection's IP.
|
||||
c.mu.Lock()
|
||||
runtime.Gosched()
|
||||
relay := relays[i%len(relays)]
|
||||
c.muInstanceURL.Lock()
|
||||
c.instanceURL = relay.url
|
||||
c.muInstanceURL.Unlock()
|
||||
c.relayConn = relay.conn
|
||||
c.mu.Unlock()
|
||||
if i == 0 {
|
||||
close(started)
|
||||
}
|
||||
}
|
||||
}()
|
||||
<-started
|
||||
|
||||
for range 1000 {
|
||||
url, ip, err := m.RelayInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
wantIP := relays[0].ip
|
||||
if url == relays[1].url.String() {
|
||||
wantIP = relays[1].ip
|
||||
}
|
||||
if !assert.Equal(t, wantIP, ip, "advertised IP must belong to relay %s", url) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_RelayInstanceAddressDisconnected(t *testing.T) {
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
client *Client
|
||||
}{
|
||||
{name: "no client"},
|
||||
{name: "not connected", client: &Client{}},
|
||||
{
|
||||
name: "closed connection",
|
||||
client: &Client{
|
||||
relayConn: stubConn{remote: staticAddr{s: "192.0.2.1:443"}},
|
||||
},
|
||||
},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
m := &Manager{relayClient: tt.client}
|
||||
url, ip, err := m.RelayInstanceAddress()
|
||||
assert.Error(t, err)
|
||||
assert.Empty(t, url, "disconnected relay must not advertise a URL")
|
||||
assert.False(t, ip.IsValid(), "disconnected relay must not advertise a stale IP")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,9 @@ package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -291,35 +293,29 @@ func TestForeignAutoClose(t *testing.T) {
|
||||
t.Fatalf("failed to serve manager: %s", err)
|
||||
}
|
||||
|
||||
// Set up a disconnect listener to track when foreign server disconnects
|
||||
foreignServerURL := toURL(srvCfg2)[0]
|
||||
disconnected := make(chan struct{})
|
||||
onDisconnect := func() {
|
||||
select {
|
||||
case disconnected <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
t.Log("open connection to another peer")
|
||||
if _, err = mgr.OpenConn(ctx, foreignServerURL, "anotherpeer", netip.Addr{}); err == nil {
|
||||
t.Fatalf("should have failed to open connection to another peer")
|
||||
}
|
||||
|
||||
// Add the disconnect listener after the connection attempt
|
||||
if err := mgr.AddCloseListener(foreignServerURL, onDisconnect); err != nil {
|
||||
t.Logf("failed to add close listener (expected if connection failed): %s", err)
|
||||
}
|
||||
|
||||
// Wait for cleanup to happen
|
||||
timeout := relayCleanupInterval + keepUnusedServerTime + 2*time.Second
|
||||
t.Logf("waiting for relay cleanup: %s", timeout)
|
||||
|
||||
select {
|
||||
case <-disconnected:
|
||||
t.Log("foreign relay connection cleaned up successfully")
|
||||
case <-time.After(timeout):
|
||||
t.Log("timeout waiting for cleanup - this might be expected if connection never established")
|
||||
deadline := time.After(timeout)
|
||||
for {
|
||||
mgr.relayClientsMutex.RLock()
|
||||
_, tracked := mgr.relayClients[foreignServerURL]
|
||||
mgr.relayClientsMutex.RUnlock()
|
||||
if !tracked {
|
||||
t.Log("foreign relay connection cleaned up successfully")
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatal("foreign relay was not cleaned up")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
t.Logf("closing manager")
|
||||
@@ -413,23 +409,24 @@ func waitForReady(ctx context.Context, m *Manager, timeout time.Duration) error
|
||||
return fmt.Errorf("manager not ready within %s", timeout)
|
||||
}
|
||||
|
||||
func TestNotifierDoubleAdd(t *testing.T) {
|
||||
func toURL(address server.ListenerConfig) []string {
|
||||
return []string{"rel://" + address.Address}
|
||||
}
|
||||
|
||||
func TestConnContextCancelledOnServerDisconnect(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
listenerCfg1 := server.ListenerConfig{
|
||||
Address: "localhost:52501",
|
||||
}
|
||||
srv, err := server.NewServer(newManagerTestServerConfig(listenerCfg1.Address))
|
||||
srvCfg := server.ListenerConfig{Address: "localhost:52601"}
|
||||
srv, err := server.NewServer(newManagerTestServerConfig(srvCfg.Address))
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create server: %s", err)
|
||||
}
|
||||
errChan := make(chan error, 1)
|
||||
go func() {
|
||||
if err := srv.Listen(listenerCfg1); err != nil {
|
||||
if err := srv.Listen(srvCfg); err != nil {
|
||||
errChan <- err
|
||||
}
|
||||
}()
|
||||
|
||||
defer func() {
|
||||
if err := srv.Shutdown(ctx); err != nil {
|
||||
t.Errorf("failed to close server: %s", err)
|
||||
@@ -440,46 +437,106 @@ func TestNotifierDoubleAdd(t *testing.T) {
|
||||
t.Fatalf("failed to start server: %s", err)
|
||||
}
|
||||
|
||||
log.Debugf("connect by alice")
|
||||
mCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
clientBob := NewManager(mCtx, toURL(listenerCfg1), "bob", iface.DefaultMTU)
|
||||
if err = clientBob.Serve(); err != nil {
|
||||
mgrBob := NewManager(mCtx, toURL(srvCfg), "bob", iface.DefaultMTU)
|
||||
if err := mgrBob.Serve(); err != nil {
|
||||
t.Fatalf("failed to serve bob manager: %s", err)
|
||||
}
|
||||
|
||||
mgr := NewManager(mCtx, toURL(srvCfg), "alice", iface.DefaultMTU)
|
||||
if err := mgr.Serve(); err != nil {
|
||||
t.Fatalf("failed to serve manager: %s", err)
|
||||
}
|
||||
|
||||
clientAlice := NewManager(mCtx, toURL(listenerCfg1), "alice", iface.DefaultMTU)
|
||||
if err = clientAlice.Serve(); err != nil {
|
||||
ra, _, err := mgr.RelayInstanceAddress()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get relay address: %s", err)
|
||||
}
|
||||
|
||||
relayedConn, err := mgr.OpenConn(ctx, ra, "bob", netip.Addr{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open conn: %s", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-relayedConn.Context().Done():
|
||||
t.Fatal("conn context cancelled while the relay is still up")
|
||||
default:
|
||||
}
|
||||
|
||||
_ = mgr.relayClient.relayConn.Close()
|
||||
|
||||
select {
|
||||
case <-relayedConn.Context().Done():
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("conn context was not cancelled after the relay connection dropped")
|
||||
}
|
||||
|
||||
if cause := context.Cause(relayedConn.Context()); !errors.Is(cause, ErrServerDisconnected) {
|
||||
t.Errorf("unexpected cancellation cause: %v, want %v", cause, ErrServerDisconnected)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnContextCauseOnLocalClose(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
srvCfg := server.ListenerConfig{Address: "localhost:52602"}
|
||||
srv, err := server.NewServer(newManagerTestServerConfig(srvCfg.Address))
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create server: %s", err)
|
||||
}
|
||||
errChan := make(chan error, 1)
|
||||
go func() {
|
||||
if err := srv.Listen(srvCfg); err != nil {
|
||||
errChan <- err
|
||||
}
|
||||
}()
|
||||
defer func() {
|
||||
if err := srv.Shutdown(ctx); err != nil {
|
||||
t.Errorf("failed to close server: %s", err)
|
||||
}
|
||||
}()
|
||||
|
||||
if err := waitForServerToStart(errChan); err != nil {
|
||||
t.Fatalf("failed to start server: %s", err)
|
||||
}
|
||||
|
||||
mCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
mgrBob := NewManager(mCtx, toURL(srvCfg), "bob", iface.DefaultMTU)
|
||||
if err := mgrBob.Serve(); err != nil {
|
||||
t.Fatalf("failed to serve bob manager: %s", err)
|
||||
}
|
||||
|
||||
mgr := NewManager(mCtx, toURL(srvCfg), "alice", iface.DefaultMTU)
|
||||
if err := mgr.Serve(); err != nil {
|
||||
t.Fatalf("failed to serve manager: %s", err)
|
||||
}
|
||||
|
||||
conn1, err := clientAlice.OpenConn(ctx, clientAlice.ServerURLs()[0], "bob", netip.Addr{})
|
||||
ra, _, err := mgr.RelayInstanceAddress()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to bind channel: %s", err)
|
||||
t.Fatalf("failed to get relay address: %s", err)
|
||||
}
|
||||
|
||||
fnCloseListener := OnServerCloseListener(func() {
|
||||
log.Infof("close listener")
|
||||
})
|
||||
|
||||
err = clientAlice.AddCloseListener(clientAlice.ServerURLs()[0], fnCloseListener)
|
||||
relayedConn, err := mgr.OpenConn(ctx, ra, "bob", netip.Addr{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to add close listener: %s", err)
|
||||
t.Fatalf("failed to open conn: %s", err)
|
||||
}
|
||||
|
||||
err = clientAlice.AddCloseListener(clientAlice.ServerURLs()[0], fnCloseListener)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to add close listener: %s", err)
|
||||
if err := relayedConn.Close(); err != nil {
|
||||
t.Fatalf("failed to close conn: %s", err)
|
||||
}
|
||||
|
||||
err = conn1.Close()
|
||||
if err != nil {
|
||||
t.Errorf("failed to close connection: %s", err)
|
||||
select {
|
||||
case <-relayedConn.Context().Done():
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("conn context was not cancelled after a local close")
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func toURL(address server.ListenerConfig) []string {
|
||||
return []string{"rel://" + address.Address}
|
||||
if cause := context.Cause(relayedConn.Context()); !errors.Is(cause, net.ErrClosed) {
|
||||
t.Errorf("unexpected cancellation cause after a local close: %v, want %v", cause, net.ErrClosed)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -63,7 +63,12 @@ func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
|
||||
if !ok {
|
||||
return nil, <-errChan
|
||||
}
|
||||
log.Infof("chosen home Relay server: %s", cr.Url)
|
||||
instanceURL, serverIP, err := cr.RelayClient.serverInstanceAddress()
|
||||
if err != nil {
|
||||
log.Infof("chosen home Relay server: %s, instance address unavailable: %v", cr.Url, err)
|
||||
return cr.RelayClient, nil
|
||||
}
|
||||
log.Infof("chosen home Relay server: %s, instance URL: %s, server IP: %s", cr.Url, instanceURL, serverIP)
|
||||
return cr.RelayClient, nil
|
||||
case <-ctx.Done():
|
||||
return nil, fmt.Errorf("connect to relay server: %w", ctx.Err())
|
||||
|
||||
Reference in New Issue
Block a user