Merge branch 'main' into refactor/relay-foreign-cache

Resolve conflicts in shared/relay/client/manager.go:

- NewManager: keep both sides. main's serverPicker.NetEvents wiring and
  this branch's ForeignRelaysStore construction are independent.
- openConnVia/openConnOnTrack: take this branch's deletion. main refined
  the RelayTrack rt.ready locking, but this branch replaces that machinery
  with the singleflight-based ForeignRelaysStore, so the refinement applies
  to a code path that no longer exists.

Carry main's netEvents plumbing into the new store so foreign relay dials
are still registered for the network change sweep.

shared/relay/client/client.go

@@ -14,6 +14,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/netbirdio/netbird/client/netevents/sweep"
 	auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
 	"github.com/netbirdio/netbird/shared/relay/client/dialer"
 	netErr "github.com/netbirdio/netbird/shared/relay/client/dialer/net"
@@ -150,6 +151,14 @@ type transportConn interface {
 	Protocol() string
 }
 
+// NetEvents is the OS network event view the relay consumes: availability
+// gating for the reconnect guard and dial registration for the network change
+// sweep.
+type NetEvents interface {
+	NetworkWatcher
+	StartDial(ctx context.Context) *sweep.Dial
+}
+
 // Client is a client for the relay server. It is responsible for establishing a connection to the relay server and
 // managing connections to other peers. All exported functions are safe to call concurrently. After close the connection,
 // the client can be reused by calling Connect again. When the client is closed, all connections are closed too.
@@ -184,6 +193,11 @@ type Client struct {
 	// datagram-sized transport is avoided on subsequent connects. Shared via
 	// the manager.
 	transportFallback *transportFallback
+
+	// netEvents registers the relay dial for the network change sweep; the
+	// read loop reports the disconnect and the guard reconnects. Shared via
+	// the manager.
+	netEvents NetEvents
 	// datagramFallbackTriggered guards a single fallback per connection so a
 	// burst of oversized datagrams triggers one reconnect, not many.
 	datagramFallbackTriggered atomic.Bool
@@ -393,6 +407,17 @@ func (c *Client) Close() error {
 }
 
 func (c *Client) connect(ctx context.Context) (*RelayAddr, error) {
+	// A sweep cancels this context, so a dial started on the old network
+	// aborts instead of waiting out its handshake timeout.
+	var dial *sweep.Dial
+	if c.netEvents != nil {
+		dial = c.netEvents.StartDial(ctx)
+	} else {
+		dial = (*sweep.Sweeper)(nil).StartDial(ctx)
+	}
+	defer dial.Release()
+	ctx = dial.Ctx()
+
 	mode := transportModeFromEnv()
 	dialers := c.getDialers(mode)
 
@@ -417,12 +442,19 @@ func (c *Client) connect(ctx context.Context) (*RelayAddr, error) {
 			return nil, fmt.Errorf("dial via FQDN: %w", err)
 		}
 	}
-	c.relayConn = conn
-	c.datagramFallbackTriggered.Store(false)
+	// Read the transport off the concrete connection: the sweeper's wrapper
+	// embeds net.Conn only, so it does not promote Protocol().
 	if tc, ok := conn.(transportConn); ok {
 		c.transport = tc.Protocol()
 	}
 
+	conn, err := dial.WrapConn(conn)
+	if err != nil {
+		return nil, fmt.Errorf("register connection: %w", err)
+	}
+	c.relayConn = conn
+	c.datagramFallbackTriggered.Store(false)
+
 	instanceURL, err := c.handShake(ctx)
 	if err != nil {
 		cErr := conn.Close()

shared/relay/client/dialer/quic/quic.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/quic-go/quic-go"
-	"github.com/quic-go/quic-go/logging"
 	log "github.com/sirupsen/logrus"
 
 	nbnet "github.com/netbirdio/netbird/client/net"
@@ -80,28 +79,6 @@ func (d Dialer) Dial(ctx context.Context, address, serverName string) (net.Conn,
 	return conn, nil
 }
 
-// connectionTracer returns a QUIC tracer that logs the DPLPMTUD result and the
-// reason a relay connection closed, so the path MTU settled on and teardown
-// cause are visible in logs. Lines carry the relay address as a structured
-// field, matching the rest of the relay client logging.
-func connectionTracer(addr string) func(context.Context, logging.Perspective, quic.ConnectionID) *logging.ConnectionTracer {
-	relayLog := log.WithField("relay", addr)
-	return func(context.Context, logging.Perspective, quic.ConnectionID) *logging.ConnectionTracer {
-		return &logging.ConnectionTracer{
-			UpdatedMTU: func(mtu logging.ByteCount, done bool) {
-				if done {
-					relayLog.Infof("QUIC path MTU settled at %d", mtu)
-					return
-				}
-				relayLog.Debugf("QUIC path MTU probing at %d", mtu)
-			},
-			ClosedConnection: func(err error) {
-				relayLog.Debugf("QUIC connection closed: %v", err)
-			},
-		}
-	}
-}
-
 func prepareURL(address string) (string, error) {
 	var host string
 	var defaultPort string

shared/relay/client/dialer/quic/quic_test.go

@@ -0,0 +1,145 @@
+package quic
+
+import (
+	"testing"
+
+	"github.com/quic-go/quic-go/qlog"
+	"github.com/quic-go/quic-go/qlogwriter"
+	log "github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
+)
+
+func TestCloseReason(t *testing.T) {
+	transportErr := qlog.TransportErrorCode(0x2) // CONNECTION_REFUSED
+	appErr := qlog.ApplicationErrorCode(42)
+
+	tests := []struct {
+		name  string
+		event qlog.ConnectionClosed
+		want  string
+	}{
+		{
+			// A close carrying nothing but an initiator still reads sensibly.
+			name:  "initiator only",
+			event: qlog.ConnectionClosed{Initiator: qlog.InitiatorLocal},
+			want:  "closed by local",
+		},
+		{
+			name: "transport error with trigger",
+			event: qlog.ConnectionClosed{
+				Initiator:       qlog.InitiatorRemote,
+				ConnectionError: &transportErr,
+				Trigger:         qlog.ConnectionCloseTriggerIdleTimeout,
+			},
+			want: "closed by remote, transport error: CONNECTION_REFUSED, trigger: idle_timeout",
+		},
+		{
+			name: "application error with reason",
+			event: qlog.ConnectionClosed{
+				Initiator:        qlog.InitiatorLocal,
+				ApplicationError: &appErr,
+				Reason:           "bye",
+			},
+			want: "closed by local, application error: 42, reason: bye",
+		},
+		{
+			// Transport and application errors are mutually exclusive in
+			// practice; if both are set the transport code wins.
+			name: "transport error takes precedence over application error",
+			event: qlog.ConnectionClosed{
+				Initiator:        qlog.InitiatorLocal,
+				ConnectionError:  &transportErr,
+				ApplicationError: &appErr,
+			},
+			want: "closed by local, transport error: CONNECTION_REFUSED",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := closeReason(tt.event); got != tt.want {
+				t.Errorf("closeReason() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestLogSinkRecordEvent(t *testing.T) {
+	tests := []struct {
+		name      string
+		event     qlogwriter.Event
+		wantLevel log.Level
+		wantMsg   string
+	}{
+		{
+			name:      "settled MTU is logged at info",
+			event:     qlog.MTUUpdated{Value: 1400, Done: true},
+			wantLevel: log.InfoLevel,
+			wantMsg:   "QUIC path MTU settled at 1400",
+		},
+		{
+			// Probing fires repeatedly during discovery, so it stays at debug.
+			name:      "MTU probe is logged at debug",
+			event:     qlog.MTUUpdated{Value: 1300, Done: false},
+			wantLevel: log.DebugLevel,
+			wantMsg:   "QUIC path MTU probing at 1300",
+		},
+		{
+			name:      "connection closed is logged at debug",
+			event:     qlog.ConnectionClosed{Initiator: qlog.InitiatorRemote},
+			wantLevel: log.DebugLevel,
+			wantMsg:   "QUIC connection closed: closed by remote",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			logger, hook := test.NewNullLogger()
+			logger.SetLevel(log.DebugLevel)
+			recorder := logSink{log: logger.WithField("relay", "relay.example.com:443")}
+
+			recorder.RecordEvent(tt.event)
+
+			entries := hook.AllEntries()
+			if len(entries) != 1 {
+				t.Fatalf("got %d log entries, want 1", len(entries))
+			}
+			if entries[0].Level != tt.wantLevel {
+				t.Errorf("level = %v, want %v", entries[0].Level, tt.wantLevel)
+			}
+			if entries[0].Message != tt.wantMsg {
+				t.Errorf("message = %q, want %q", entries[0].Message, tt.wantMsg)
+			}
+			if relay := entries[0].Data["relay"]; relay != "relay.example.com:443" {
+				t.Errorf("relay field = %v, want relay.example.com:443", relay)
+			}
+		})
+	}
+}
+
+// Events the relay client does not care about must not produce log lines.
+func TestLogSinkIgnoresUnhandledEvents(t *testing.T) {
+	logger, hook := test.NewNullLogger()
+	logger.SetLevel(log.DebugLevel)
+	recorder := logSink{log: logger.WithField("relay", "relay.example.com:443")}
+
+	recorder.RecordEvent(qlog.PacketLost{})
+
+	if entries := hook.AllEntries(); len(entries) != 0 {
+		t.Errorf("got %d log entries, want 0", len(entries))
+	}
+}
+
+func TestLogSinkSupportsSchemas(t *testing.T) {
+	trace := logSink{log: log.WithField("relay", "relay.example.com:443")}
+
+	if !trace.SupportsSchemas(qlog.EventSchema) {
+		t.Errorf("SupportsSchemas(%q) = false, want true", qlog.EventSchema)
+	}
+	if trace.SupportsSchemas("urn:ietf:params:qlog:events:http3-12") {
+		t.Error("SupportsSchemas() = true for an unrelated schema, want false")
+	}
+	if trace.AddProducer() == nil {
+		t.Error("AddProducer() = nil, want a recorder")
+	}
+}

shared/relay/client/dialer/quic/tracer.go

@@ -0,0 +1,70 @@
+package quic
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/quic-go/quic-go"
+	"github.com/quic-go/quic-go/qlog"
+	"github.com/quic-go/quic-go/qlogwriter"
+	log "github.com/sirupsen/logrus"
+)
+
+// logSink implements both qlogwriter.Trace and qlogwriter.Recorder, forwarding
+// the few qlog events the relay client cares about to logrus instead of
+// writing a qlog file. It holds no mutable state and logrus entries are safe
+// to share, so one value can serve every producer on the connection.
+type logSink struct {
+	log *log.Entry
+}
+
+func (s logSink) AddProducer() qlogwriter.Recorder { return s }
+
+func (s logSink) SupportsSchemas(schema string) bool { return schema == qlog.EventSchema }
+
+func (s logSink) RecordEvent(event qlogwriter.Event) {
+	switch e := event.(type) {
+	case qlog.MTUUpdated:
+		if e.Done {
+			s.log.Infof("QUIC path MTU settled at %d", e.Value)
+			return
+		}
+		s.log.Debugf("QUIC path MTU probing at %d", e.Value)
+	case qlog.ConnectionClosed:
+		s.log.Debugf("QUIC connection closed: %s", closeReason(e))
+	}
+}
+
+func (s logSink) Close() error { return nil }
+
+// connectionTracer returns a QUIC tracer that logs the DPLPMTUD result and the
+// reason a relay connection closed, so the path MTU settled on and teardown
+// cause are visible in logs. Lines carry the relay address as a structured
+// field, matching the rest of the relay client logging.
+func connectionTracer(addr string) func(context.Context, bool, quic.ConnectionID) qlogwriter.Trace {
+	relayLog := log.WithField("relay", addr)
+	return func(context.Context, bool, quic.ConnectionID) qlogwriter.Trace {
+		return logSink{log: relayLog}
+	}
+}
+
+// closeReason renders a ConnectionClosed event as a single line. The event
+// carries the error as separate initiator, code, trigger and reason fields,
+// any of which may be unset.
+func closeReason(e qlog.ConnectionClosed) string {
+	parts := []string{fmt.Sprintf("closed by %s", e.Initiator)}
+	switch {
+	case e.ConnectionError != nil:
+		parts = append(parts, fmt.Sprintf("transport error: %s", *e.ConnectionError))
+	case e.ApplicationError != nil:
+		parts = append(parts, fmt.Sprintf("application error: %d", *e.ApplicationError))
+	}
+	if e.Trigger != "" {
+		parts = append(parts, fmt.Sprintf("trigger: %s", e.Trigger))
+	}
+	if e.Reason != "" {
+		parts = append(parts, fmt.Sprintf("reason: %s", e.Reason))
+	}
+	return strings.Join(parts, ", ")
+}

shared/relay/client/fallback_opener_test.go

@@ -73,7 +73,7 @@ func TestHandleResult_PreferredFailsStartsOther(t *testing.T) {
 	// The fallback attempt opens against a stalling listener so startOther's
 	// goroutine blocks on Connect until raceCtx is cancelled by t.Cleanup.
 	serverAddr, _ := stallingRelayListener(t)
-	c.opener.foreignStore = NewForeignRelaysStore(c.raceCtx, hmacTokenStore, "alice", 1280, newTransportFallback(), func(string) {}, keepUnusedServerTime)
+	c.opener.foreignStore = NewForeignRelaysStore(c.raceCtx, hmacTokenStore, "alice", 1280, newTransportFallback(), nil, func(string) {}, keepUnusedServerTime)
 	c.remoteRelayServer = RelayServer{Addr: serverAddr}
 	c.preferForeign = false
 

shared/relay/client/foreign_relays_store.go

@@ -29,11 +29,12 @@ type ForeignRelaysStore struct {
 	peerID               string
 	mtu                  uint16
 	transportFallback    *transportFallback
+	netEvents            NetEvents
 	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 {
+func NewForeignRelaysStore(ctx context.Context, tokenStore *relayAuth.TokenStore, peerID string, mtu uint16, transportFallback *transportFallback, netEvents NetEvents, onDisconnect func(string), keepUnusedServerTime time.Duration) *ForeignRelaysStore {
 	return &ForeignRelaysStore{
 		clients:              make(map[string]*foreignRelay),
 		ctx:                  ctx,
@@ -41,6 +42,7 @@ func NewForeignRelaysStore(ctx context.Context, tokenStore *relayAuth.TokenStore
 		peerID:               peerID,
 		mtu:                  mtu,
 		transportFallback:    transportFallback,
+		netEvents:            netEvents,
 		onDisconnect:         onDisconnect,
 		keepUnusedServerTime: keepUnusedServerTime,
 	}
@@ -75,6 +77,7 @@ func (f *ForeignRelaysStore) acquire(remoteRelayServer RelayServer) (*foreignRel
 
 		relayClient := NewClientWithServerIP(remoteRelayServer.Addr, remoteRelayServer.IP, f.tokenStore, f.peerID, f.mtu)
 		relayClient.SetTransportFallback(f.transportFallback)
+		relayClient.netEvents = f.netEvents
 		if err := relayClient.Connect(f.ctx); err != nil {
 			return nil, err
 		}

shared/relay/client/foreign_relays_store_test.go

@@ -11,7 +11,7 @@ import (
 
 func newTestForeignStore(t *testing.T, ctx context.Context) *ForeignRelaysStore {
 	t.Helper()
-	return NewForeignRelaysStore(ctx, hmacTokenStore, "alice", 1280, newTransportFallback(), func(string) {}, keepUnusedServerTime)
+	return NewForeignRelaysStore(ctx, hmacTokenStore, "alice", 1280, newTransportFallback(), nil, func(string) {}, keepUnusedServerTime)
 }
 
 func TestForeignStore_AcquireDedupsConcurrentOpens(t *testing.T) {

shared/relay/client/guard.go

@@ -9,7 +9,25 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-const defaultMaxBackoffInterval = 60 * time.Second
+const (
+	defaultMaxBackoffInterval = 60 * time.Second
+
+	// quickReconnectBudget bounds how long a quick reconnect waits for the
+	// network before handing the retry over to the ticker.
+	quickReconnectBudget = 1500 * time.Millisecond
+
+	// verdictSettleWindow is how long an online verdict must hold before it
+	// is trusted: the disconnect often precedes the OS offline flag by a few
+	// milliseconds.
+	verdictSettleWindow = 200 * time.Millisecond
+)
+
+// NetworkWatcher is the availability view the guard gates reconnects on.
+type NetworkWatcher interface {
+	Wait(ctx context.Context) (bool, error)
+	IsOnline() bool
+	WaitSettled(ctx context.Context, budget, settleWindow time.Duration) bool
+}
 
 // Guard manage the reconnection tries to the Relay server in case of disconnection event.
 type Guard struct {
@@ -22,6 +40,9 @@ type Guard struct {
 	// attempts.
 	maxBackoffInterval time.Duration
 
+	// netWatcher gates reconnect attempts on OS-reported network availability.
+	netWatcher NetworkWatcher
+
 	// lastErr is the error from the most recent failed reconnect attempt,
 	// surfaced as the home relay status while disconnected.
 	lastErr atomic.Pointer[error]
@@ -29,7 +50,7 @@ type Guard struct {
 
 // NewGuard creates a new guard for the relay client. A non-positive
 // maxBackoffInterval falls back to defaultMaxBackoffInterval.
-func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard {
+func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netWatcher NetworkWatcher) *Guard {
 	if maxBackoffInterval <= 0 {
 		maxBackoffInterval = defaultMaxBackoffInterval
 	}
@@ -38,6 +59,7 @@ func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard {
 		OnReconnected:      make(chan struct{}, 1),
 		serverPicker:       sp,
 		maxBackoffInterval: maxBackoffInterval,
+		netWatcher:         netWatcher,
 	}
 	return g
 }
@@ -70,11 +92,23 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
 
 	// start a ticker to pick a new server
 	ticker := g.exponentTicker(ctx)
-	defer ticker.Stop()
+	defer func() {
+		ticker.Stop()
+	}()
 
 	for {
 		select {
 		case <-ticker.C:
+			// suspend reconnect attempts while the OS reports no usable network
+			if g.netWatcher != nil {
+				if waited, err := g.netWatcher.Wait(ctx); err != nil {
+					return
+				} else if waited {
+					ticker.Stop()
+					ticker = g.exponentTicker(ctx)
+					continue
+				}
+			}
 			if err := g.retry(ctx); err != nil {
 				log.Errorf("failed to pick new Relay server: %s", err)
 				g.setLastError(err)
@@ -100,8 +134,18 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool
 		return false
 	}
 
-	if cancelled := waiteBeforeRetry(parentCtx); !cancelled {
-		return false
+	if g.netWatcher != nil {
+		if ok := g.netWatcher.WaitSettled(parentCtx, quickReconnectBudget, verdictSettleWindow); !ok {
+			return false
+		}
+		// Still offline after the budget: leave the retry to the ticker.
+		if !g.netWatcher.IsOnline() {
+			return false
+		}
+	} else {
+		if cancelled := waitBeforeRetry(parentCtx); !cancelled {
+			return false
+		}
 	}
 
 	log.Infof("try to reconnect to Relay server: %s", rc.connectionURL)
@@ -156,16 +200,18 @@ func (g *Guard) notifyReconnected() {
 func (g *Guard) exponentTicker(ctx context.Context) *backoff.Ticker {
 	bo := backoff.WithContext(&backoff.ExponentialBackOff{
 		InitialInterval: 2 * time.Second,
-		Multiplier:      2,
-		MaxInterval:     g.maxBackoffInterval,
-		Clock:           backoff.SystemClock,
+		// Spreads the reconnects of every client that lost the same relay server.
+		RandomizationFactor: backoff.DefaultRandomizationFactor,
+		Multiplier:          2,
+		MaxInterval:         g.maxBackoffInterval,
+		Clock:               backoff.SystemClock,
 	}, ctx)
 
 	return backoff.NewTicker(bo)
 }
 
-func waiteBeforeRetry(ctx context.Context) bool {
-	timer := time.NewTimer(1500 * time.Millisecond)
+func waitBeforeRetry(ctx context.Context) bool {
+	timer := time.NewTimer(quickReconnectBudget)
 	defer timer.Stop()
 
 	select {

shared/relay/client/manager.go

@@ -49,6 +49,11 @@ func WithMaxBackoffInterval(d time.Duration) ManagerOption {
 	return func(m *Manager) { m.maxBackoffInterval = d }
 }
 
+// WithNetEvents injects the OS network event handling.
+func WithNetEvents(events NetEvents) ManagerOption {
+	return func(m *Manager) { m.netEvents = events }
+}
+
 // Manager is a manager for the relay client instances. It establishes one persistent connection to the given relay URL
 // and automatically reconnect to them in case disconnection.
 // The manager also manage temporary relay connection. If a client wants to communicate with a client on a
@@ -75,6 +80,7 @@ type Manager struct {
 
 	mtu                uint16
 	maxBackoffInterval time.Duration
+	netEvents          NetEvents
 
 	cleanupInterval      time.Duration
 	keepUnusedServerTime time.Duration
@@ -110,9 +116,10 @@ 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.NetEvents = m.netEvents
+	m.foreign = NewForeignRelaysStore(ctx, tokenStore, peerID, mtu, tf, m.netEvents, m.onServerDisconnected, m.keepUnusedServerTime)
 	m.serverPicker.ServerURLs.Store(serverURLs)
-	m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval)
+	m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netEvents)
 	return m
 }
 

shared/relay/client/picker.go

@@ -30,6 +30,7 @@ type ServerPicker struct {
 	MTU               uint16
 	ConnectionTimeout time.Duration
 	TransportFallback *transportFallback
+	NetEvents         NetEvents
 }
 
 func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
@@ -73,6 +74,7 @@ func (sp *ServerPicker) startConnection(ctx context.Context, resultChan chan con
 	log.Infof("try to connecting to relay server: %s", url)
 	relayClient := NewClient(url, sp.TokenStore, sp.PeerID, sp.MTU)
 	relayClient.SetTransportFallback(sp.TransportFallback)
+	relayClient.netEvents = sp.NetEvents
 	err := relayClient.Connect(ctx)
 	resultChan <- connResult{
 		RelayClient: relayClient,