Merge remote-tracking branch 'origin/main' into dmitri-event-aggregation

Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>
2026-09-09 00:11:28 +02:00 · 2026-06-22 10:57:11 +02:00
parent 17cc13f20f c9e99659ea
commit 4a1341883b
119 changed files with 3952 additions and 1899 deletions
--- a/shared/relay/client/client.go
+++ b/shared/relay/client/client.go
@@ -145,6 +145,11 @@ func (cc *connContainer) close() {
 	}
 }

+// transportConn is implemented by relay connections that know their transport.
+type transportConn interface {
+	Protocol() string
+}
+
 // 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.
@@ -182,6 +187,18 @@ type Client struct {
 	// datagramFallbackTriggered guards a single fallback per connection so a
 	// burst of oversized datagrams triggers one reconnect, not many.
 	datagramFallbackTriggered atomic.Bool
+
+	// transport is the negotiated relay transport of the
+	// current connection, guarded by mu.
+	transport string
+}
+
+// Transport returns the negotiated relay transport of the current connection,
+// or an empty string when not connected.
+func (c *Client) Transport() string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.transport
 }

 // SetTransportFallback wires the shared datagram-transport fallback tracker.
@@ -402,6 +419,9 @@ func (c *Client) connect(ctx context.Context) (*RelayAddr, error) {
 	}
 	c.relayConn = conn
 	c.datagramFallbackTriggered.Store(false)
+	if tc, ok := conn.(transportConn); ok {
+		c.transport = tc.Protocol()
+	}

 	instanceURL, err := c.handShake(ctx)
 	if err != nil {
@@ -792,6 +812,7 @@ func (c *Client) close(gracefullyExit bool) error {
 		return nil
 	}
 	c.serviceIsRunning = false
+	c.transport = ""

 	c.muInstanceURL.Lock()
 	c.instanceURL = nil
--- a/shared/relay/client/dialer/quic/conn.go
+++ b/shared/relay/client/dialer/quic/conn.go
@@ -57,6 +57,11 @@ func (c *Conn) Write(b []byte) (int, error) {
 	return len(b), nil
 }

+// Protocol returns the transport name for this connection.
+func (c *Conn) Protocol() string {
+	return Network
+}
+
 func (c *Conn) RemoteAddr() net.Addr {
 	return c.session.RemoteAddr()
 }
--- a/shared/relay/client/dialer/quic/quic.go
+++ b/shared/relay/client/dialer/quic/quic.go
@@ -59,14 +59,12 @@ func (d Dialer) Dial(ctx context.Context, address, serverName string) (net.Conn,

 	udpConn, err := nbnet.ListenUDP("udp", &net.UDPAddr{Port: 0})
 	if err != nil {
-		log.Errorf("failed to listen on UDP: %s", err)
-		return nil, err
+		return nil, fmt.Errorf("listen udp: %w", err)
 	}

 	udpAddr, err := net.ResolveUDPAddr("udp", quicURL)
 	if err != nil {
-		log.Errorf("failed to resolve UDP address: %s", err)
-		return nil, err
+		return nil, fmt.Errorf("resolve %s: %w", quicURL, err)
 	}

 	session, err := quic.Dial(ctx, udpConn, udpAddr, tlsClientConfig, quicConfig)
@@ -74,7 +72,7 @@ func (d Dialer) Dial(ctx context.Context, address, serverName string) (net.Conn,
 		if errors.Is(err, context.Canceled) {
 			return nil, err
 		}
-		log.Errorf("failed to dial to Relay server via QUIC '%s': %s", quicURL, err)
+		log.Debugf("failed to dial to Relay server via QUIC '%s': %s", quicURL, err)
 		return nil, err
 	}

--- a/shared/relay/client/dialer/race_dialer.go
+++ b/shared/relay/client/dialer/race_dialer.go
@@ -3,6 +3,7 @@ package dialer
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net"
 	"time"

@@ -71,6 +72,7 @@ func (r *RaceDial) Dial(ctx context.Context) (net.Conn, error) {

 	connChan := make(chan dialResult, len(r.dialerFns))
 	winnerConn := make(chan net.Conn, 1)
+	errChan := make(chan error, 1)
 	abortCtx, abort := context.WithCancel(ctx)
 	defer abort()

@@ -78,11 +80,11 @@ func (r *RaceDial) Dial(ctx context.Context) (net.Conn, error) {
 		go r.dial(dfn, abortCtx, connChan)
 	}

-	go r.processResults(connChan, winnerConn, abort)
+	go r.processResults(connChan, winnerConn, errChan, abort)

 	conn, ok := <-winnerConn
 	if !ok {
-		return nil, errors.New("failed to dial to Relay server on any protocol")
+		return nil, <-errChan
 	}
 	return conn, nil
 }
@@ -90,6 +92,7 @@ func (r *RaceDial) Dial(ctx context.Context) (net.Conn, error) {
 // dialSequential tries each dialer in order, returning the first connection and
 // falling back to the next on failure.
 func (r *RaceDial) dialSequential(ctx context.Context) (net.Conn, error) {
+	var errs []error
 	for _, dfn := range r.dialerFns {
 		if err := ctx.Err(); err != nil {
 			return nil, err
@@ -103,12 +106,13 @@ func (r *RaceDial) dialSequential(ctx context.Context) (net.Conn, error) {
 				return nil, err
 			}
 			r.log.Errorf("failed to dial via %s: %s", dfn.Protocol(), err)
+			errs = append(errs, fmt.Errorf("%s: %w", dfn.Protocol(), err))
 			continue
 		}
 		r.log.Infof("successfully dialed via: %s", dfn.Protocol())
 		return conn, nil
 	}
-	return nil, errors.New("failed to dial to Relay server on any protocol")
+	return nil, dialErr(errs)
 }

 func (r *RaceDial) dial(dfn DialeFn, abortCtx context.Context, connChan chan dialResult) {
@@ -120,8 +124,9 @@ func (r *RaceDial) dial(dfn DialeFn, abortCtx context.Context, connChan chan dia
 	connChan <- dialResult{Conn: conn, Protocol: dfn.Protocol(), Err: err}
 }

-func (r *RaceDial) processResults(connChan chan dialResult, winnerConn chan net.Conn, abort context.CancelFunc) {
+func (r *RaceDial) processResults(connChan chan dialResult, winnerConn chan net.Conn, errChan chan error, abort context.CancelFunc) {
 	var hasWinner bool
+	errsByProtocol := make(map[string]error)
 	for i := 0; i < len(r.dialerFns); i++ {
 		dr := <-connChan
 		if dr.Err != nil {
@@ -129,6 +134,7 @@ func (r *RaceDial) processResults(connChan chan dialResult, winnerConn chan net.
 				r.log.Infof("connection attempt aborted via: %s", dr.Protocol)
 			} else {
 				r.log.Errorf("failed to dial via %s: %s", dr.Protocol, dr.Err)
+				errsByProtocol[dr.Protocol] = fmt.Errorf("%s: %w", dr.Protocol, dr.Err)
 			}
 			continue
 		}
@@ -146,5 +152,29 @@ func (r *RaceDial) processResults(connChan chan dialResult, winnerConn chan net.
 		hasWinner = true
 		winnerConn <- dr.Conn
 	}
+	if !hasWinner {
+		errChan <- dialErr(r.orderedErrs(errsByProtocol))
+	}
 	close(winnerConn)
 }
+
+// orderedErrs returns the per-protocol errors in dialer order, so the combined
+// error is stable regardless of which attempt failed first.
+func (r *RaceDial) orderedErrs(byProtocol map[string]error) []error {
+	errs := make([]error, 0, len(byProtocol))
+	for _, dfn := range r.dialerFns {
+		if err, ok := byProtocol[dfn.Protocol()]; ok {
+			errs = append(errs, err)
+		}
+	}
+	return errs
+}
+
+// dialErr combines per-dialer failures, preserving the underlying reasons
+// (e.g. "connection refused") rather than a generic message.
+func dialErr(errs []error) error {
+	if len(errs) == 0 {
+		return errors.New("no relay transport available")
+	}
+	return errors.Join(errs...)
+}
--- a/shared/relay/client/dialer/ws/conn.go
+++ b/shared/relay/client/dialer/ws/conn.go
@@ -33,6 +33,11 @@ func NewConn(wsConn *websocket.Conn, serverAddress string, underlying net.Conn)
 	}
 }

+// Protocol returns the transport name for this connection.
+func (c *Conn) Protocol() string {
+	return Network
+}
+
 func (c *Conn) Read(b []byte) (n int, err error) {
 	t, ioReader, err := c.Conn.Reader(c.ctx)
 	if err != nil {
--- a/shared/relay/client/dialer/ws/ws.go
+++ b/shared/relay/client/dialer/ws/ws.go
@@ -22,7 +22,7 @@ type Dialer struct {
 }

 func (d Dialer) Protocol() string {
-	return "WS"
+	return Network
 }

 func (d Dialer) Dial(ctx context.Context, address, serverName string) (net.Conn, error) {
@@ -39,7 +39,12 @@ func (d Dialer) Dial(ctx context.Context, address, serverName string) (net.Conn,
 		if errors.Is(err, context.Canceled) {
 			return nil, err
 		}
-		log.Errorf("failed to dial to Relay server '%s': %s", wsURL, err)
+		// websocket.Dial wraps the cause in verbose layers; surface the
+		// underlying network error when present.
+		var opErr *net.OpError
+		if errors.As(err, &opErr) {
+			return nil, opErr
+		}
 		return nil, err
 	}
 	if resp.Body != nil {
--- a/shared/relay/client/dialers_generic_test.go
+++ b/shared/relay/client/dialers_generic_test.go
@@ -41,14 +41,14 @@ func TestGetDialers(t *testing.T) {
 		preferWS bool
 		want     []string
 	}{
-		{name: "auto races quic and ws", mode: "auto", mtu: iface.DefaultMTU, want: []string{"quic", "WS"}},
-		{name: "ws pinned", mode: "ws", mtu: iface.DefaultMTU, want: []string{"WS"}},
+		{name: "auto races quic and ws", mode: "auto", mtu: iface.DefaultMTU, want: []string{"quic", "ws"}},
+		{name: "ws pinned", mode: "ws", mtu: iface.DefaultMTU, want: []string{"ws"}},
 		{name: "quic pinned", mode: "quic", mtu: iface.DefaultMTU, want: []string{"quic"}},
-		{name: "prefer-quic orders quic first", mode: "prefer-quic", mtu: iface.DefaultMTU, want: []string{"quic", "WS"}},
-		{name: "prefer-ws orders ws first", mode: "prefer-ws", mtu: iface.DefaultMTU, want: []string{"WS", "quic"}},
-		{name: "mtu above default forces ws", mode: "auto", mtu: iface.DefaultMTU + 100, want: []string{"WS"}},
-		{name: "sticky fallback forces ws in auto", mode: "auto", mtu: iface.DefaultMTU, preferWS: true, want: []string{"WS"}},
-		{name: "sticky fallback forces ws in prefer-quic", mode: "prefer-quic", mtu: iface.DefaultMTU, preferWS: true, want: []string{"WS"}},
+		{name: "prefer-quic orders quic first", mode: "prefer-quic", mtu: iface.DefaultMTU, want: []string{"quic", "ws"}},
+		{name: "prefer-ws orders ws first", mode: "prefer-ws", mtu: iface.DefaultMTU, want: []string{"ws", "quic"}},
+		{name: "mtu above default forces ws", mode: "auto", mtu: iface.DefaultMTU + 100, want: []string{"ws"}},
+		{name: "sticky fallback forces ws in auto", mode: "auto", mtu: iface.DefaultMTU, preferWS: true, want: []string{"ws"}},
+		{name: "sticky fallback forces ws in prefer-quic", mode: "prefer-quic", mtu: iface.DefaultMTU, preferWS: true, want: []string{"ws"}},
 		{name: "quic pin overrides sticky fallback", mode: "quic", mtu: iface.DefaultMTU, preferWS: true, want: []string{"quic"}},
 	}

@@ -91,11 +91,11 @@ func TestStickyFallbackAfterDatagramTooLarge(t *testing.T) {
 	}

 	// First dial races both transports.
-	assert.Equal(t, []string{"quic", "WS"}, protocols(c.getDialers(transportModeFromEnv())))
+	assert.Equal(t, []string{"quic", "ws"}, protocols(c.getDialers(transportModeFromEnv())))

 	// An oversized datagram records the fallback for this server.
 	c.onDatagramTooLarge(&closeTrackingConn{}, netErr.ErrDatagramTooLarge)

 	// The reconnect now sticks to WebSocket.
-	assert.Equal(t, []string{"WS"}, protocols(c.getDialers(transportModeFromEnv())))
+	assert.Equal(t, []string{"ws"}, protocols(c.getDialers(transportModeFromEnv())))
 }
--- a/shared/relay/client/guard.go
+++ b/shared/relay/client/guard.go
@@ -2,6 +2,7 @@ package client

 import (
 	"context"
+	"sync/atomic"
 	"time"

 	"github.com/cenkalti/backoff/v4"
@@ -20,6 +21,10 @@ type Guard struct {
 	// maxBackoffInterval caps the exponential backoff between reconnect
 	// attempts.
 	maxBackoffInterval time.Duration
+
+	// lastErr is the error from the most recent failed reconnect attempt,
+	// surfaced as the home relay status while disconnected.
+	lastErr atomic.Pointer[error]
 }

 // NewGuard creates a new guard for the relay client. A non-positive
@@ -37,6 +42,15 @@ func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard {
 	return g
 }

+// LastError returns the error from the most recent failed reconnect attempt, or
+// nil if reconnection last succeeded.
+func (g *Guard) LastError() error {
+	if p := g.lastErr.Load(); p != nil {
+		return *p
+	}
+	return nil
+}
+
 // StartReconnectTrys is called when the relay client is disconnected from the relay server.
 // It attempts to reconnect to the relay server. The function first tries a quick reconnect
 // to the same server that was used before, if the server URL is still valid. If the quick
@@ -63,6 +77,7 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
 		case <-ticker.C:
 			if err := g.retry(ctx); err != nil {
 				log.Errorf("failed to pick new Relay server: %s", err)
+				g.setLastError(err)
 				continue
 			}
 			return
@@ -72,6 +87,10 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
 	}
 }

+func (g *Guard) setLastError(err error) {
+	g.lastErr.Store(&err)
+}
+
 func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool {
 	if rc == nil {
 		return false
@@ -89,6 +108,7 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool

 	if err := rc.Connect(parentCtx); err != nil {
 		log.Errorf("failed to reconnect to relay server: %s", err)
+		g.setLastError(err)
 		return false
 	}
 	return true
@@ -100,6 +120,7 @@ func (g *Guard) retry(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	g.setLastError(nil)

 	// prevent to work with a deprecated Relay client instance
 	g.drainRelayClientChan()
@@ -125,6 +146,7 @@ func (g *Guard) isServerURLStillValid(rc *Client) bool {
 }

 func (g *Guard) notifyReconnected() {
+	g.setLastError(nil)
 	select {
 	case g.OnReconnected <- struct{}{}:
 	default:
--- a/shared/relay/client/manager.go
+++ b/shared/relay/client/manager.go
@@ -43,6 +43,17 @@ type OnServerCloseListener func()
 // ManagerOption configures a Manager at construction time.
 type ManagerOption func(*Manager)

+// RelayConnState is the connection state of a single relay server.
+type RelayConnState struct {
+	// URL is the server's instance address when connected, otherwise the
+	// configured server URL.
+	URL string
+	// Transport is the negotiated transport, empty if not connected.
+	Transport string
+	// Err is set when the relay is not connected.
+	Err error
+}
+
 // 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 {
@@ -130,6 +141,9 @@ func (m *Manager) Serve() error {

 	client, err := m.serverPicker.PickServer(m.ctx)
 	if err != nil {
+		// record the initial failure so status shows the real reason before
+		// the guard's first retry tick
+		m.reconnectGuard.setLastError(err)
 		go m.reconnectGuard.StartReconnectTrys(m.ctx, nil)
 	} else {
 		m.storeClient(client)
@@ -242,6 +256,56 @@ func (m *Manager) ServerURLs() []string {
 	return m.serverPicker.ServerURLs.Load().([]string)
 }

+// RelayConnectError returns the error from the most recent failed home relay
+// reconnect attempt, or nil if the relay last connected successfully.
+func (m *Manager) RelayConnectError() error {
+	return m.reconnectGuard.LastError()
+}
+
+// RelayStates returns the connection state of the home relay and every foreign
+// relay the manager currently tracks.
+func (m *Manager) RelayStates() []RelayConnState {
+	var states []RelayConnState
+
+	m.relayClientMu.RLock()
+	home := m.relayClient
+	m.relayClientMu.RUnlock()
+	if home != nil {
+		st := relayConnState(home)
+		// The home relay reconnects through the guard, so the real failure
+		// reason lives there rather than on the (stale) client.
+		if st.Err != nil {
+			if gErr := m.reconnectGuard.LastError(); gErr != nil {
+				st.Err = gErr
+			}
+		}
+		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))
+		}
+	}
+
+	return states
+}
+
 // HasRelayAddress returns true if the manager is serving. With this method can check if the peer can communicate with
 // Relay service.
 func (m *Manager) HasRelayAddress() bool {
@@ -460,3 +524,11 @@ func (m *Manager) notifyOnDisconnectListeners(serverAddress string) {
 	}
 	delete(m.onDisconnectedListeners, serverAddress)
 }
+
+func relayConnState(c *Client) RelayConnState {
+	addr, err := c.ServerInstanceURL()
+	if err != nil {
+		return RelayConnState{URL: c.connectionURL, Err: err}
+	}
+	return RelayConnState{URL: addr, Transport: c.Transport()}
+}
--- a/shared/relay/client/picker.go
+++ b/shared/relay/client/picker.go
@@ -40,6 +40,7 @@ func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {

 	connResultChan := make(chan connResult, totalServers)
 	successChan := make(chan connResult, 1)
+	errChan := make(chan error, 1)
 	concurrentLimiter := make(chan struct{}, maxConcurrentServers)

 	log.Debugf("pick server from list: %v", sp.ServerURLs.Load().([]string))
@@ -54,17 +55,17 @@ func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
 		}(url)
 	}

-	go sp.processConnResults(connResultChan, successChan)
+	go sp.processConnResults(connResultChan, successChan, errChan)

 	select {
 	case cr, ok := <-successChan:
 		if !ok {
-			return nil, errors.New("failed to connect to any relay server: all attempts failed")
+			return nil, <-errChan
 		}
 		log.Infof("chosen home Relay server: %s", cr.Url)
 		return cr.RelayClient, nil
 	case <-ctx.Done():
-		return nil, fmt.Errorf("failed to connect to any relay server: %w", ctx.Err())
+		return nil, fmt.Errorf("connect to relay server: %w", ctx.Err())
 	}
 }

@@ -80,12 +81,14 @@ func (sp *ServerPicker) startConnection(ctx context.Context, resultChan chan con
 	}
 }

-func (sp *ServerPicker) processConnResults(resultChan chan connResult, successChan chan connResult) {
+func (sp *ServerPicker) processConnResults(resultChan chan connResult, successChan chan connResult, errChan chan error) {
 	var hasSuccess bool
+	var errs []error
 	for numOfResults := 0; numOfResults < cap(resultChan); numOfResults++ {
 		cr := <-resultChan
 		if cr.Err != nil {
 			log.Tracef("failed to connect to Relay server: %s: %v", cr.Url, cr.Err)
+			errs = append(errs, cr.Err)
 			continue
 		}
 		log.Infof("connected to Relay server: %s", cr.Url)
@@ -101,5 +104,16 @@ func (sp *ServerPicker) processConnResults(resultChan chan connResult, successCh
 		hasSuccess = true
 		successChan <- cr
 	}
+	if !hasSuccess {
+		errChan <- pickErr(errs)
+	}
 	close(successChan)
 }
+
+// pickErr combines per-server connection failures into a single error.
+func pickErr(errs []error) error {
+	if len(errs) == 0 {
+		return errors.New("no relay server available")
+	}
+	return errors.Join(errs...)
+}
--- a/shared/signal/client/grpc.go
+++ b/shared/signal/client/grpc.go
@@ -2,9 +2,11 @@ package client

 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
 	"time"

 	"github.com/cenkalti/backoff/v4"
@@ -23,7 +25,23 @@ import (
 	"github.com/netbirdio/netbird/util/wsproxy"
 )

-const healthCheckTimeout = 5 * time.Second
+const (
+	// receiveInactivityThreshold is how long the receive stream may be silent
+	// before the watchdog actively probes it. The gRPC transport can stay
+	// healthy (keepalive satisfied) while the server stops delivering messages,
+	// which the transport layer cannot detect.
+	receiveInactivityThreshold = 30 * time.Second
+	// receiveProbeTimeout is how long the watchdog waits for its self-addressed
+	// probe to round-trip back on the stream before declaring the receive
+	// direction dead.
+	receiveProbeTimeout = 10 * time.Second
+	// receiveWatchdogInterval is how often the watchdog evaluates the stream.
+	receiveWatchdogInterval = 10 * time.Second
+)
+
+// errReceiveStreamStalled is reported when the receive stream is transport-alive
+// but no longer delivering messages, so the stream is torn down to reconnect.
+var errReceiveStreamStalled = errors.New("signal receive stream stalled")

 // ConnStateNotifier is a wrapper interface of the status recorder
 type ConnStateNotifier interface {
@@ -52,6 +70,14 @@ type GrpcClient struct {
 	decryptionWorker       *Worker
 	decryptionWorkerCancel context.CancelFunc
 	decryptionWg           sync.WaitGroup
+
+	// lastReceived holds the Unix-nano timestamp of the last message read from
+	// the receive stream, used by the receive watchdog.
+	lastReceived atomic.Int64
+	// receiveStalled is set by the receive watchdog when the stream is
+	// transport-alive but no longer delivering messages. It is the source of
+	// truth IsHealthy reads, and is cleared once any frame is received again.
+	receiveStalled atomic.Bool
 }

 // NewClient creates a new Signal client
@@ -148,9 +174,9 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes

 		// connect to Signal stream identifying ourselves with a public WireGuard key
 		// todo once the key rotation logic has been implemented, consider changing to some other identifier (received from management)
-		ctx, cancelStream := context.WithCancel(ctx)
+		streamCtx, cancelStream := context.WithCancel(ctx)
 		defer cancelStream()
-		stream, err := c.connect(ctx, c.key.PublicKey().String())
+		stream, err := c.connect(streamCtx, c.key.PublicKey().String())
 		if err != nil {
 			log.Warnf("disconnected from the Signal Exchange due to an error: %v", err)
 			return err
@@ -164,9 +190,16 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes
 		// Start worker pool if not already started
 		c.startEncryptionWorker(msgHandler)

+		// Guard the receive direction: the transport can stay healthy while the
+		// server stops delivering messages. The watchdog reconnects via cancelStream.
+		c.markReceived()
+		go c.watchReceiveStream(streamCtx, cancelStream)
+
 		// start receiving messages from the Signal stream (from other peers through signal)
 		err = c.receive(stream)
 		if err != nil {
+			// Check the parent context, not streamCtx: a watchdog-triggered
+			// cancelStream must reconnect, only a parent cancel is shutdown.
 			if ctx.Err() != nil {
 				log.Debugf("signal connection context has been canceled, this usually indicates shutdown")
 				return nil
@@ -252,7 +285,10 @@ func (c *GrpcClient) Ready() bool {
 	return c.signalConn.GetState() == connectivity.Ready || c.signalConn.GetState() == connectivity.Idle
 }

-// IsHealthy probes the gRPC connection and returns false on errors
+// IsHealthy reports whether the Signal connection is usable, based on the
+// transport state plus the receive watchdog's verdict, and updates the status
+// recorder accordingly. It does not actively probe: the watchdog
+// (watchReceiveStream) owns probing the receive path and reconnecting.
 func (c *GrpcClient) IsHealthy() bool {
 	switch c.signalConn.GetState() {
 	case connectivity.TransientFailure:
@@ -265,16 +301,8 @@ func (c *GrpcClient) IsHealthy() bool {
 	case connectivity.Ready:
 	}

-	ctx, cancel := context.WithTimeout(c.ctx, healthCheckTimeout)
-	defer cancel()
-	_, err := c.realClient.Send(ctx, &proto.EncryptedMessage{
-		Key:       c.key.PublicKey().String(),
-		RemoteKey: "dummy",
-		Body:      nil,
-	})
-	if err != nil {
-		c.notifyDisconnected(err)
-		log.Warnf("health check returned: %s", err)
+	if c.receiveStalled.Load() {
+		c.notifyDisconnected(errReceiveStreamStalled)
 		return false
 	}
 	c.notifyConnected()
@@ -398,6 +426,68 @@ func (c *GrpcClient) Send(msg *proto.Message) error {
 	return err
 }

+// markReceived records that a frame was just read from the receive stream and
+// clears the stalled flag.
+func (c *GrpcClient) markReceived() {
+	c.lastReceived.Store(time.Now().UnixNano())
+	c.receiveStalled.Store(false)
+}
+
+// idleSinceReceive returns how long the receive stream has been silent.
+func (c *GrpcClient) idleSinceReceive() time.Duration {
+	return time.Since(time.Unix(0, c.lastReceived.Load()))
+}
+
+// watchReceiveStream guards against a receive stream that is transport-alive but
+// no longer delivering messages. While the stream is idle past
+// receiveInactivityThreshold it sends a self-addressed probe that the Signal
+// server routes back to this client. If the probe does not round-trip within
+// receiveProbeTimeout the receive direction is considered dead and cancelStream
+// is called so the retry loop reconnects.
+func (c *GrpcClient) watchReceiveStream(ctx context.Context, cancelStream context.CancelFunc) {
+	ticker := time.NewTicker(receiveWatchdogInterval)
+	defer ticker.Stop()
+
+	var probeSentAt time.Time
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if c.idleSinceReceive() < receiveInactivityThreshold {
+				probeSentAt = time.Time{}
+				continue
+			}
+
+			if !probeSentAt.IsZero() && time.Since(probeSentAt) >= receiveProbeTimeout {
+				log.Warnf("signal receive stream stalled: no messages for %s and probe did not return, reconnecting", c.idleSinceReceive().Round(time.Second))
+				c.receiveStalled.Store(true)
+				c.notifyDisconnected(errReceiveStreamStalled)
+				cancelStream()
+				return
+			}
+
+			if probeSentAt.IsZero() {
+				if err := c.sendReceiveProbe(); err != nil {
+					log.Debugf("failed to send signal receive probe: %v", err)
+				}
+				probeSentAt = time.Now()
+			}
+		}
+	}
+}
+
+// sendReceiveProbe sends a self-addressed heartbeat. The Signal server routes it
+// back to this client, exercising the exact receive path the watchdog guards.
+func (c *GrpcClient) sendReceiveProbe() error {
+	self := c.key.PublicKey().String()
+	return c.Send(&proto.Message{
+		Key:       self,
+		RemoteKey: self,
+		Body:      &proto.Body{Type: proto.Body_HEARTBEAT},
+	})
+}
+
 // receive receives messages from other peers coming through the Signal Exchange
 // and distributes them to worker threads for processing
 func (c *GrpcClient) receive(stream proto.SignalExchange_ConnectStreamClient) error {
@@ -419,6 +509,9 @@ func (c *GrpcClient) receive(stream proto.SignalExchange_ConnectStreamClient) er
 			return err
 		}

+		// Any frame from the server proves the receive direction is alive.
+		c.markReceived()
+
 		if msg == nil {
 			continue
 		}
--- a/shared/signal/client/watchdog_test.go
+++ b/shared/signal/client/watchdog_test.go
@@ -0,0 +1,84 @@
+package client
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"google.golang.org/grpc"
+
+	sigProto "github.com/netbirdio/netbird/shared/signal/proto"
+	"github.com/netbirdio/netbird/signal/server"
+)
+
+func startTestSignalServer(t *testing.T) string {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	s := grpc.NewServer()
+	srv, err := server.NewServer(context.Background(), otel.Meter(""))
+	require.NoError(t, err)
+	sigProto.RegisterSignalExchangeServer(s, srv)
+
+	go func() {
+		_ = s.Serve(lis)
+	}()
+	t.Cleanup(s.Stop)
+
+	return lis.Addr().String()
+}
+
+// TestReceiveProbeRoundTrips verifies that the watchdog's self-addressed heartbeat
+// is routed back to the same client through the signal server. This round-trip is
+// what lets the watchdog confirm the receive direction is still delivering.
+func TestReceiveProbeRoundTrips(t *testing.T) {
+	addr := startTestSignalServer(t)
+
+	key, err := wgtypes.GenerateKey()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	client, err := NewClient(ctx, addr, key, false)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = client.Close() })
+
+	received := make(chan struct{}, 1)
+	go func() {
+		_ = client.Receive(ctx, func(msg *sigProto.Message) error {
+			if msg.GetBody().GetType() == sigProto.Body_HEARTBEAT && msg.GetKey() == key.PublicKey().String() {
+				select {
+				case received <- struct{}{}:
+				default:
+				}
+			}
+			return nil
+		})
+	}()
+
+	streamReady := make(chan struct{})
+	go func() {
+		client.WaitStreamConnected()
+		close(streamReady)
+	}()
+	select {
+	case <-streamReady:
+	case <-time.After(5 * time.Second):
+		t.Fatal("signal stream did not connect within timeout")
+	}
+
+	require.NoError(t, client.sendReceiveProbe())
+
+	select {
+	case <-received:
+	case <-time.After(3 * time.Second):
+		t.Fatal("self-addressed heartbeat did not round-trip back through the signal server")
+	}
+}
--- a/shared/signal/proto/signalexchange.pb.go
+++ b/shared/signal/proto/signalexchange.pb.go
@@ -30,6 +30,7 @@ const (
 	Body_CANDIDATE Body_Type = 2
 	Body_MODE      Body_Type = 4
 	Body_GO_IDLE   Body_Type = 5
+	Body_HEARTBEAT Body_Type = 6
 )

 // Enum value maps for Body_Type.
@@ -40,6 +41,7 @@ var (
 		2: "CANDIDATE",
 		4: "MODE",
 		5: "GO_IDLE",
+		6: "HEARTBEAT",
 	}
 	Body_Type_value = map[string]int32{
 		"OFFER":     0,
@@ -47,6 +49,7 @@ var (
 		"CANDIDATE": 2,
 		"MODE":      4,
 		"GO_IDLE":   5,
+		"HEARTBEAT": 6,
 	}
 )

@@ -463,7 +466,7 @@ var file_signalexchange_proto_rawDesc = []byte{
 	0x52, 0x09, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x28, 0x0a, 0x04, 0x62,
 	0x6f, 0x64, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x73, 0x69, 0x67, 0x6e,
 	0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x42, 0x6f, 0x64, 0x79, 0x52,
-	0x04, 0x62, 0x6f, 0x64, 0x79, 0x22, 0xc3, 0x04, 0x0a, 0x04, 0x42, 0x6f, 0x64, 0x79, 0x12, 0x2d,
+	0x04, 0x62, 0x6f, 0x64, 0x79, 0x22, 0xd2, 0x04, 0x0a, 0x04, 0x42, 0x6f, 0x64, 0x79, 0x12, 0x2d,
 	0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x73,
 	0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x42, 0x6f,
 	0x64, 0x79, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x18, 0x0a,
@@ -491,38 +494,39 @@ var file_signalexchange_proto_rawDesc = []byte{
 	0x52, 0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x88, 0x01, 0x01, 0x12, 0x29,
 	0x0a, 0x0d, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x18,
 	0x0b, 0x20, 0x01, 0x28, 0x0c, 0x48, 0x02, 0x52, 0x0d, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65,
-	0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x88, 0x01, 0x01, 0x22, 0x43, 0x0a, 0x04, 0x54, 0x79, 0x70,
+	0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x88, 0x01, 0x01, 0x22, 0x52, 0x0a, 0x04, 0x54, 0x79, 0x70,
 	0x65, 0x12, 0x09, 0x0a, 0x05, 0x4f, 0x46, 0x46, 0x45, 0x52, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06,
 	0x41, 0x4e, 0x53, 0x57, 0x45, 0x52, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09, 0x43, 0x41, 0x4e, 0x44,
 	0x49, 0x44, 0x41, 0x54, 0x45, 0x10, 0x02, 0x12, 0x08, 0x0a, 0x04, 0x4d, 0x4f, 0x44, 0x45, 0x10,
-	0x04, 0x12, 0x0b, 0x0a, 0x07, 0x47, 0x4f, 0x5f, 0x49, 0x44, 0x4c, 0x45, 0x10, 0x05, 0x42, 0x15,
-	0x0a, 0x13, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64,
-	0x64, 0x72, 0x65, 0x73, 0x73, 0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f,
-	0x6e, 0x49, 0x64, 0x42, 0x10, 0x0a, 0x0e, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72,
-	0x76, 0x65, 0x72, 0x49, 0x50, 0x4a, 0x04, 0x08, 0x09, 0x10, 0x0a, 0x22, 0x2e, 0x0a, 0x04, 0x4d,
-	0x6f, 0x64, 0x65, 0x12, 0x1b, 0x0a, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x08, 0x48, 0x00, 0x52, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x88, 0x01, 0x01,
-	0x42, 0x09, 0x0a, 0x07, 0x5f, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x22, 0x6d, 0x0a, 0x0f, 0x52,
-	0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x28,
-	0x0a, 0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x50, 0x75, 0x62, 0x4b, 0x65,
-	0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61,
-	0x73, 0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x6f, 0x73, 0x65,
-	0x6e, 0x70, 0x61, 0x73, 0x73, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73,
-	0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x32, 0xb9, 0x01, 0x0a, 0x0e, 0x53,
-	0x69, 0x67, 0x6e, 0x61, 0x6c, 0x45, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x4c, 0x0a,
-	0x04, 0x53, 0x65, 0x6e, 0x64, 0x12, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78,
-	0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64,
-	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c,
-	0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74,
-	0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x12, 0x59, 0x0a, 0x0d, 0x43,
-	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x20, 0x2e, 0x73,
-	0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e,
-	0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20,
-	0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e,
-	0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
-	0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x04, 0x12, 0x0b, 0x0a, 0x07, 0x47, 0x4f, 0x5f, 0x49, 0x44, 0x4c, 0x45, 0x10, 0x05, 0x12, 0x0d,
+	0x0a, 0x09, 0x48, 0x45, 0x41, 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x06, 0x42, 0x15, 0x0a,
+	0x13, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e,
+	0x49, 0x64, 0x42, 0x10, 0x0a, 0x0e, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76,
+	0x65, 0x72, 0x49, 0x50, 0x4a, 0x04, 0x08, 0x09, 0x10, 0x0a, 0x22, 0x2e, 0x0a, 0x04, 0x4d, 0x6f,
+	0x64, 0x65, 0x12, 0x1b, 0x0a, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x18, 0x01, 0x20, 0x01,
+	0x28, 0x08, 0x48, 0x00, 0x52, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x88, 0x01, 0x01, 0x42,
+	0x09, 0x0a, 0x07, 0x5f, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x22, 0x6d, 0x0a, 0x0f, 0x52, 0x6f,
+	0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x28, 0x0a,
+	0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73,
+	0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e,
+	0x70, 0x61, 0x73, 0x73, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x53,
+	0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x32, 0xb9, 0x01, 0x0a, 0x0e, 0x53, 0x69,
+	0x67, 0x6e, 0x61, 0x6c, 0x45, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x4c, 0x0a, 0x04,
+	0x53, 0x65, 0x6e, 0x64, 0x12, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63,
+	0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d,
+	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65,
+	0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65,
+	0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x12, 0x59, 0x0a, 0x0d, 0x43, 0x6f,
+	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x20, 0x2e, 0x73, 0x69,
+	0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63,
+	0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e,
+	0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45,
+	0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22,
+	0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }

 var (
--- a/shared/signal/proto/signalexchange.proto
+++ b/shared/signal/proto/signalexchange.proto
@@ -48,6 +48,7 @@ message Body {
    CANDIDATE = 2;
    MODE = 4;
    GO_IDLE = 5;
+    HEARTBEAT = 6;
  }
  Type type = 1;
  string payload = 2;