sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-09-01 20:41:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

Bridges embed to use the supervisor + remove providing of establishedChan from caller

Browse Source 
(We now do AsyncStart + waitEstablishedOrDone, so everything is managed inside the
supervisor)
This commit is contained in:
riccardom
2026-06-18 17:10:17 +02:00
parent b39e9df194
commit 9628f016da
6 changed files with 78 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
client/cmd/up.go
View File

@@ -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 {

26
client/embed/embed.go
View File

@@ -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

50
client/internal/connect.go
View File

@@ -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()

15
client/internal/connect_android_default.go
View File

@@ -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,
)
}
}

17
client/internal/connect_android_embed.go
View File

@@ -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)
}

55
client/internal/connect_lifecycle.go
View File

@@ -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