[client] Fix metrics push getting stuck off after engine restart

Engine restarts (backoff retries within the same login session) cancel
e.ctx, which the push goroutine's lifetime was tied to. The goroutine
died silently but ClientMetrics.push stayed non-nil since only an
explicit stop clears it, so the next UpdatePushFromMgm call saw a
"push already running" state and never restarted it.

Give the Engine its own metricsCtx sourced from ConnectClient.ctx,
which outlives engine restarts, so handleMetricsUpdate stops tying the
push to the wrong-scoped context. Additionally make ClientMetrics.push
an atomic.Pointer that the push goroutine clears via CompareAndSwap on
exit, so the tracked state can never drift from the goroutine's actual
lifetime regardless of which context a future caller passes in.

client/internal/connect.go

@@ -403,6 +403,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
 			StateManager:   stateManager,
 			UpdateManager:  c.updateManager,
 			ClientMetrics:  c.clientMetrics,
+			MetricsCtx:     c.ctx,
 		}, mobileDependency)
 		engine.SetSyncResponsePersistence(c.persistSyncResponse)
 		c.engine = engine

client/internal/engine.go

@@ -172,6 +172,7 @@ type EngineServices struct {
 	StateManager   *statemanager.Manager
 	UpdateManager  *updater.Manager
 	ClientMetrics  *metrics.ClientMetrics
+	MetricsCtx     context.Context
 }
 
 // Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
@@ -264,6 +265,7 @@ type Engine struct {
 
 	// clientMetrics collects and pushes metrics
 	clientMetrics *metrics.ClientMetrics
+	metricsCtx    context.Context
 
 	jobExecutor   *jobexec.Executor
 	jobExecutorWG sync.WaitGroup
@@ -316,6 +318,7 @@ func NewEngine(
 		probeStunTurn:      relay.NewStunTurnProbe(relay.DefaultCacheTTL),
 		jobExecutor:        jobexec.NewExecutor(),
 		clientMetrics:      services.ClientMetrics,
+		metricsCtx:         services.MetricsCtx,
 		updateManager:      services.UpdateManager,
 		syncStoreDir:       config.StateDir,
 	}
@@ -1073,7 +1076,7 @@ func (e *Engine) handleMetricsUpdate(config *mgmProto.MetricsConfig) {
 		return
 	}
 	log.Infof("received metrics configuration from management: enabled=%v", config.GetEnabled())
-	e.clientMetrics.UpdatePushFromMgm(e.ctx, config.GetEnabled())
+	e.clientMetrics.UpdatePushFromMgm(e.metricsCtx, config.GetEnabled())
 }
 
 func toFlowLoggerConfig(config *mgmProto.FlowConfig) (*nftypes.FlowConfig, error) {

client/internal/metrics/metrics.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -75,7 +76,7 @@ type ClientMetrics struct {
 	agentInfo AgentInfo
 	mu        sync.RWMutex
 
-	push       *Push
+	push       atomic.Pointer[Push]
 	pushMu     sync.Mutex
 	wg         sync.WaitGroup
 	pushCancel context.CancelFunc
@@ -167,10 +168,7 @@ func (c *ClientMetrics) UpdateAgentInfo(agentInfo AgentInfo, publicKey string) {
 	c.agentInfo = agentInfo
 	c.mu.Unlock()
 
-	c.pushMu.Lock()
-	push := c.push
-	c.pushMu.Unlock()
-	if push != nil {
+	if push := c.push.Load(); push != nil {
 		push.SetPeerID(agentInfo.peerID)
 	}
 }
@@ -194,7 +192,7 @@ func (c *ClientMetrics) StartPush(ctx context.Context, config PushConfig) {
 	c.pushMu.Lock()
 	defer c.pushMu.Unlock()
 
-	if c.push != nil {
+	if c.push.Load() != nil {
 		log.Warnf("metrics push already running")
 		return
 	}
@@ -230,13 +228,13 @@ func (c *ClientMetrics) UpdatePushFromMgm(ctx context.Context, enabled bool) {
 	defer c.pushMu.Unlock()
 
 	if enabled {
-		if c.push != nil {
+		if c.push.Load() != nil {
 			return
 		}
 		log.Infof("enabled metrics push by management")
 		c.startPushLocked(ctx, PushConfigFromEnv())
 	} else {
-		if c.push == nil {
+		if c.push.Load() == nil {
 			return
 		}
 		log.Infof("disabled metrics push by management")
@@ -261,22 +259,23 @@ func (c *ClientMetrics) startPushLocked(ctx context.Context, config PushConfig)
 
 	ctx, cancel := context.WithCancel(ctx)
 	c.pushCancel = cancel
+	c.push.Store(push)
 
 	c.wg.Add(1)
 	go func() {
 		defer c.wg.Done()
 		push.Start(ctx)
+		c.push.CompareAndSwap(push, nil)
 	}()
-	c.push = push
 }
 
 // stopPushLocked stops push. Caller must hold pushMu.
 func (c *ClientMetrics) stopPushLocked() {
-	if c.push == nil {
+	if c.push.Load() == nil {
 		return
 	}
 
 	c.pushCancel()
 	c.wg.Wait()
-	c.push = nil
+	c.push.Store(nil)
 }