diff --git a/client/cmd/up.go b/client/cmd/up.go index be1c06cc1..fb7213d64 100644 --- a/client/cmd/up.go +++ b/client/cmd/up.go @@ -204,7 +204,7 @@ func runInForegroundMode(ctx context.Context, cmd *cobra.Command, activeProf *pr connectClient := internal.NewConnectClient(ctx, r) SetupDebugHandler(ctx, config, r, connectClient, "") - return connectClient.Run(config, nil, nil, util.FindFirstLogPath(logFiles)) + return connectClient.Run(config, nil, util.FindFirstLogPath(logFiles)) } func runInDaemonMode(ctx context.Context, cmd *cobra.Command, pm *profilemanager.ProfileManager, activeProf *profilemanager.Profile, profileSwitched bool) error { diff --git a/client/embed/embed.go b/client/embed/embed.go index 13e7ebfbf..e3b8af8ca 100644 --- a/client/embed/embed.go +++ b/client/embed/embed.go @@ -267,29 +267,21 @@ func (c *Client) Start(startCtx context.Context) error { client := internal.NewConnectClient(ctx, c.recorder) client.SetSyncResponsePersistence(true) - // either startup error (permanent backoff err) or nil err (successful engine up) + // The supervisor owns the run; we wait until it is established, ends with a + // startup error (permanent backoff err), or startCtx expires. // TODO: make after-startup backoff err available - run := make(chan struct{}) - clientErr := make(chan error, 1) - go func() { - if err := client.Run(c.config, nil, run, ""); err != nil { - clientErr <- err - } - }() + client.RunAsync(c.config, nil) - select { - case <-startCtx.Done(): - // Cancel the client context before stopping: Engine.Start blocks on the - // signal stream while holding the engine mutex and only unblocks on - // cancellation. Stopping first would deadlock on that mutex. + if err := client.WaitEstablishedOrDone(startCtx); err != nil { + // Either startCtx expired while connecting, or the run ended before it + // established. Cancel the client context before stopping: Engine.Start + // blocks on the signal stream while holding the engine mutex and only + // unblocks on cancellation. Stopping first would deadlock on that mutex. cancel() if stopErr := client.Stop(); stopErr != nil { - return fmt.Errorf("stop error after context done. Stop error: %w. Context done: %w", stopErr, startCtx.Err()) + return fmt.Errorf("stop error after startup failure. Stop error: %w. Startup: %w", stopErr, err) } - return startCtx.Err() - case err := <-clientErr: return fmt.Errorf("startup: %w", err) - case <-run: } c.connect = client diff --git a/client/internal/connect.go b/client/internal/connect.go index 3b313a7dd..04bb85c36 100644 --- a/client/internal/connect.go +++ b/client/internal/connect.go @@ -49,9 +49,20 @@ import ( "github.com/netbirdio/netbird/version" ) -// androidRunOverride is set on Android to inject mobile dependencies -// when using embed.Client (which calls Run() with empty MobileDependency). -var androidRunOverride func(c *ConnectClient, config *profilemanager.Config, connEstablishedChan chan struct{}, logPath string) error +// androidMobileDep is set on Android to inject the MobileDependency for runs +// started through the generic entry points (Run/RunAsync, e.g. embed.Client). +// nil on other platforms, where the dependency is empty. +var androidMobileDep func(config *profilemanager.Config) MobileDependency + +// mobileDependency returns the MobileDependency for a run started via the +// generic entry points. On Android the androidMobileDep provider supplies +// platform stubs (or real implementations); elsewhere it is empty. +func (c *ConnectClient) mobileDependency(config *profilemanager.Config) MobileDependency { + if androidMobileDep != nil { + return androidMobileDep(config) + } + return MobileDependency{} +} type ConnectClient struct { ctx context.Context @@ -88,20 +99,16 @@ func (c *ConnectClient) SetUpdateManager(um *updater.Manager) { // Run with main logic. md carries optional gRPC metadata (e.g. the UI // user-agent) to forward to the management/signal services; nil when none. -func (c *ConnectClient) Run(config *profilemanager.Config, md metadata.MD, connEstablishedChan chan struct{}, logPath string) error { - if androidRunOverride != nil { - return androidRunOverride(c, config, connEstablishedChan, logPath) - } - return c.sup.start(config, md, MobileDependency{}, connEstablishedChan, logPath) +func (c *ConnectClient) Run(config *profilemanager.Config, md metadata.MD, logPath string) error { + return c.sup.start(config, md, c.mobileDependency(config), logPath) } -// RunAsync starts a client run without blocking. Used by the daemon, which -// drives the lifecycle through the supervisor rather than blocking on Run; it -// then waits for the outcome via WaitEstablishedOrDone. The run's lifecycle +// RunAsync starts a client run without blocking. Used by the daemon and embed, +// which drive the lifecycle through the supervisor rather than blocking on Run; +// they then wait for the outcome via WaitEstablishedOrDone. The run's lifecycle // channels are created and owned by the supervisor — callers never hold them. func (c *ConnectClient) RunAsync(config *profilemanager.Config, md metadata.MD) { - est := make(chan struct{}) - c.sup.startAsync(config, md, MobileDependency{}, est, "", nil) + c.sup.startAsync(config, md, c.mobileDependency(config), "", nil) } // WaitEstablishedOrDone blocks until the in-flight run becomes established (nil), @@ -133,7 +140,7 @@ func (c *ConnectClient) RunOnAndroid( StateFilePath: stateFilePath, TempDir: cacheDir, } - return c.sup.start(config, nil, mobileDependency, nil, "") + return c.sup.start(config, nil, mobileDependency, "") } func (c *ConnectClient) RunOniOS( @@ -155,7 +162,7 @@ func (c *ConnectClient) RunOniOS( StateFilePath: stateFilePath, TempDir: cacheDir, } - return c.sup.start(config, nil, mobileDependency, nil, logFilePath) + return c.sup.start(config, nil, mobileDependency, logFilePath) } // run executes a single client run. runCtx is owned by the supervisor: cancelling @@ -423,12 +430,13 @@ func (c *ConnectClient) run(runCtx context.Context, config *profilemanager.Confi log.Infof("Netbird engine started, the IP is: %s", peerConfig.GetAddress()) state.Set(StatusConnected) - if connEstablishedChan != nil { - select { - case <-connEstablishedChan: - default: - close(connEstablishedChan) - } + // The supervisor owns connEstablishedChan and it is always present. Guard + // against a double close: operation re-runs on ErrResetConnection retries + // within the same run, and the channel is closed only on the first connect. + select { + case <-connEstablishedChan: + default: + close(connEstablishedChan) } <-engineCtx.Done() diff --git a/client/internal/connect_android_default.go b/client/internal/connect_android_default.go index 518867270..6c6780d78 100644 --- a/client/internal/connect_android_default.go +++ b/client/internal/connect_android_default.go @@ -60,20 +60,17 @@ var _ listener.NetworkChangeListener = noopNetworkChangeListener{} var _ dns.ReadyListener = noopDnsReadyListener{} func init() { - // Wire up the default override so embed.Client.Start() works on Android - // with netstack mode. Provides complete no-op stubs for all mobile + // Wire up the default MobileDependency provider so embed.Client.Start() works + // on Android with netstack mode. Provides complete no-op stubs for all mobile // dependencies so the engine's existing Android code paths work unchanged. - // Applications that need P2P ICE or real DNS should replace this by - // setting androidRunOverride before calling Start(). - androidRunOverride = func(c *ConnectClient, config *profilemanager.Config, connEstablishedChan chan struct{}, logPath string) error { - return c.runOnAndroidEmbed( - config, + // Applications that need P2P ICE or real DNS should replace this by setting + // androidMobileDep before calling Start(). + androidMobileDep = func(config *profilemanager.Config) MobileDependency { + return mobileDependencyForEmbed( noopIFaceDiscover{}, noopNetworkChangeListener{}, []netip.AddrPort{}, noopDnsReadyListener{}, - connEstablishedChan, - logPath, ) } } diff --git a/client/internal/connect_android_embed.go b/client/internal/connect_android_embed.go index 32c2c7449..747bd2de5 100644 --- a/client/internal/connect_android_embed.go +++ b/client/internal/connect_android_embed.go @@ -7,28 +7,21 @@ import ( "github.com/netbirdio/netbird/client/internal/dns" "github.com/netbirdio/netbird/client/internal/listener" - "github.com/netbirdio/netbird/client/internal/profilemanager" "github.com/netbirdio/netbird/client/internal/stdnet" ) -// runOnAndroidEmbed is like RunOnAndroid but accepts a connEstablishedChan -// so embed.Client.Start() can detect when the engine is ready. -// It provides complete MobileDependency so the engine's existing -// Android code paths work unchanged. -func (c *ConnectClient) runOnAndroidEmbed( - config *profilemanager.Config, +// mobileDependencyForEmbed builds the MobileDependency used by embed.Client on +// Android so the engine's existing Android code paths work unchanged. +func mobileDependencyForEmbed( iFaceDiscover stdnet.ExternalIFaceDiscover, networkChangeListener listener.NetworkChangeListener, dnsAddresses []netip.AddrPort, dnsReadyListener dns.ReadyListener, - connEstablishedChan chan struct{}, - logPath string, -) error { - mobileDependency := MobileDependency{ +) MobileDependency { + return MobileDependency{ IFaceDiscover: iFaceDiscover, NetworkChangeListener: networkChangeListener, HostDNSAddresses: dnsAddresses, DnsReadyListener: dnsReadyListener, } - return c.sup.start(config, nil, mobileDependency, connEstablishedChan, logPath) } diff --git a/client/internal/connect_lifecycle.go b/client/internal/connect_lifecycle.go index 6a3acc71d..8bf3cc983 100644 --- a/client/internal/connect_lifecycle.go +++ b/client/internal/connect_lifecycle.go @@ -42,28 +42,29 @@ const ( // // reply is used only by opStatus. waitCtx is used only by opWaitEstablished. type lifecycleCmd struct { - op lifecycleOp - config *profilemanager.Config - md metadata.MD - mobileDep MobileDependency - connEstablishedChan chan struct{} - logPath string - done chan error - reply chan bool - waitCtx context.Context + op lifecycleOp + config *profilemanager.Config + md metadata.MD + mobileDep MobileDependency + logPath string + done chan error + reply chan bool + waitCtx context.Context } -// runState holds the per-run termination signal owned by the loop goroutine. It -// never escapes the supervisor as an API; the only readers are the per-wait -// goroutines the loop spawns for opWaitEstablished. +// runState holds the lifecycle channels of a single in-flight run, owned by the +// loop goroutine. It never escapes the supervisor as an API; the only readers +// are the per-wait goroutines the loop spawns for opWaitEstablished. // -// ended is closed (broadcast) when the run terminates, so any number of waiters -// can observe it; err is the run's end result, valid only after ended is closed. -// The "established" signal is not duplicated here — it is the start command's -// connEstablishedChan, snapshotted directly from curStart when a waiter needs it. +// connEstablishedChan is closed by the run once the connection is established. +// The supervisor creates and owns it — callers no longer supply it; they observe +// it through waitEstablishedOrDone. ended is closed (broadcast) when the run +// terminates, so any number of waiters can observe it; err is the run's end +// result, valid only after ended is closed. type runState struct { - ended chan struct{} // closed by finishRun when the run terminates - err error // run end result, valid after ended is closed + connEstablishedChan chan struct{} // closed by the run on established + ended chan struct{} // closed by finishRun when the run terminates + err error // run end result, valid after ended is closed } // runEndResult is sent by the run goroutine to the supervisor when a run ends, @@ -145,12 +146,12 @@ func (s *supervisor) handleStart(cmd lifecycleCmd) { } s.runCancel = cancel s.curStart = &cmd - s.curRun = &runState{ended: make(chan struct{})} + s.curRun = &runState{connEstablishedChan: make(chan struct{}), ended: make(chan struct{})} - go func(ctx context.Context, cfg *profilemanager.Config, m MobileDependency, rc chan struct{}, lp string) { - err := s.run(ctx, cfg, m, rc, lp) + go func(ctx context.Context, cfg *profilemanager.Config, m MobileDependency, established chan struct{}, lp string) { + err := s.run(ctx, cfg, m, established, lp) s.runEnded <- runEndResult{err: err} - }(runCtx, cmd.config, cmd.mobileDep, cmd.connEstablishedChan, cmd.logPath) + }(runCtx, cmd.config, cmd.mobileDep, s.curRun.connEstablishedChan, cmd.logPath) } func (s *supervisor) handleStop(cmd lifecycleCmd) { @@ -194,7 +195,7 @@ func (s *supervisor) handleWaitEstablished(cmd lifecycleCmd) { return } rs := s.curRun - established := s.curStart.connEstablishedChan + established := rs.connEstablishedChan ctx := cmd.waitCtx go func() { select { @@ -233,8 +234,8 @@ func (s *supervisor) shutdown() { // startAsync enqueues a start without blocking. If done is non-nil it receives // the run's end result (or errAlreadyRunning on rejection, or the context error // on shutdown). -func (s *supervisor) startAsync(config *profilemanager.Config, md metadata.MD, mobileDep MobileDependency, connEstablishedChan chan struct{}, logPath string, done chan error) { - cmd := lifecycleCmd{op: opStart, config: config, md: md, mobileDep: mobileDep, connEstablishedChan: connEstablishedChan, logPath: logPath, done: done} +func (s *supervisor) startAsync(config *profilemanager.Config, md metadata.MD, mobileDep MobileDependency, logPath string, done chan error) { + cmd := lifecycleCmd{op: opStart, config: config, md: md, mobileDep: mobileDep, logPath: logPath, done: done} select { case s.cmdCh <- cmd: case <-s.ctx.Done(): @@ -244,9 +245,9 @@ func (s *supervisor) startAsync(config *profilemanager.Config, md metadata.MD, m // start enqueues a start and blocks until the run terminates, preserving the // blocking contract of the legacy Run entry points. -func (s *supervisor) start(config *profilemanager.Config, md metadata.MD, mobileDep MobileDependency, connEstablishedChan chan struct{}, logPath string) error { +func (s *supervisor) start(config *profilemanager.Config, md metadata.MD, mobileDep MobileDependency, logPath string) error { done := make(chan error, 1) - s.startAsync(config, md, mobileDep, connEstablishedChan, logPath, done) + s.startAsync(config, md, mobileDep, logPath, done) select { case err := <-done: return err