From 0503a18644163cf3a7af1f66ae56da882a1acf96 Mon Sep 17 00:00:00 2001 From: riccardom Date: Tue, 16 Jun 2026 16:01:12 +0200 Subject: [PATCH] Wraps client lifetime into a supervisor - define needed supervisor context/variables - will use runCancel as knob to know if the client is running. No extra boolean flags - runWaiter is used to signal to the async run caller --- client/internal/connect.go | 35 +++- client/internal/connect_android_embed.go | 2 +- client/internal/connect_lifecycle.go | 201 +++++++++++++++++++++++ 3 files changed, 229 insertions(+), 9 deletions(-) create mode 100644 client/internal/connect_lifecycle.go diff --git a/client/internal/connect.go b/client/internal/connect.go index e38bc2f58..b20bff289 100644 --- a/client/internal/connect.go +++ b/client/internal/connect.go @@ -63,6 +63,10 @@ type ConnectClient struct { updateManager *updater.Manager persistSyncResponse bool + + // sup serializes all start/stop requests so two lifecycle operations can + // never overlap. See connect_lifecycle.go. + sup *supervisor } func NewConnectClient( @@ -70,12 +74,14 @@ func NewConnectClient( config *profilemanager.Config, statusRecorder *peer.Status, ) *ConnectClient { - return &ConnectClient{ + c := &ConnectClient{ ctx: ctx, config: config, statusRecorder: statusRecorder, engineMutex: sync.Mutex{}, } + c.sup = newSupervisor(ctx, c.run) + return c } func (c *ConnectClient) SetUpdateManager(um *updater.Manager) { @@ -87,7 +93,7 @@ func (c *ConnectClient) Run(runningChan chan struct{}, logPath string) error { if androidRunOverride != nil { return androidRunOverride(c, runningChan, logPath) } - return c.run(MobileDependency{}, runningChan, logPath) + return c.sup.start(MobileDependency{}, runningChan, logPath) } // RunOnAndroid with main logic on mobile system @@ -110,7 +116,7 @@ func (c *ConnectClient) RunOnAndroid( StateFilePath: stateFilePath, TempDir: cacheDir, } - return c.run(mobileDependency, nil, "") + return c.sup.start(mobileDependency, nil, "") } func (c *ConnectClient) RunOniOS( @@ -128,10 +134,12 @@ func (c *ConnectClient) RunOniOS( DnsManager: dnsManager, StateFilePath: stateFilePath, } - return c.run(mobileDependency, nil, "") + return c.sup.start(mobileDependency, nil, "") } -func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan struct{}, logPath string) error { +// run executes a single client run. runCtx is owned by the supervisor: cancelling +// it tears the run down (it is the parent of the per-attempt engine context). +func (c *ConnectClient) run(runCtx context.Context, mobileDependency MobileDependency, runningChan chan struct{}, logPath string) error { defer func() { if r := recover(); r != nil { rec := c.statusRecorder @@ -240,13 +248,13 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan defer c.statusRecorder.ClientStop() operation := func() error { // if context cancelled we not start new backoff cycle - if c.ctx.Err() != nil { + if runCtx.Err() != nil { return nil } state.Set(StatusConnecting) - engineCtx, cancel := context.WithCancel(c.ctx) + engineCtx, cancel := context.WithCancel(runCtx) defer func() { _, err := state.Status() c.statusRecorder.MarkManagementDisconnected(err) @@ -435,7 +443,9 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan log.Debugf("exiting client retry loop due to unrecoverable error: %s", err) if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) { state.Set(StatusNeedsLogin) - _ = c.Stop() + // Called from inside the run goroutine: tear the engine down + // directly, never through the lifecycle queue (would deadlock). + _ = c.stopEngine() } return err } @@ -512,7 +522,16 @@ func (c *ConnectClient) Status() StatusType { return status } +// Stop serializes a stop request through the lifecycle supervisor and blocks +// until the in-flight run is fully torn down. func (c *ConnectClient) Stop() error { + return c.sup.stop() +} + +// stopEngine stops the engine directly, bypassing the lifecycle queue. It is +// only safe to call from inside the run goroutine itself (which is what the +// supervisor is waiting on); routing it through the queue there would deadlock. +func (c *ConnectClient) stopEngine() error { engine := c.Engine() if engine != nil { if err := engine.Stop(); err != nil { diff --git a/client/internal/connect_android_embed.go b/client/internal/connect_android_embed.go index 18f72e841..c96a0b41d 100644 --- a/client/internal/connect_android_embed.go +++ b/client/internal/connect_android_embed.go @@ -28,5 +28,5 @@ func (c *ConnectClient) runOnAndroidEmbed( HostDNSAddresses: dnsAddresses, DnsReadyListener: dnsReadyListener, } - return c.run(mobileDependency, runningChan, logPath) + return c.sup.start(mobileDependency, runningChan, logPath) } diff --git a/client/internal/connect_lifecycle.go b/client/internal/connect_lifecycle.go new file mode 100644 index 000000000..78bff5b92 --- /dev/null +++ b/client/internal/connect_lifecycle.go @@ -0,0 +1,201 @@ +package internal + +import ( + "context" + "errors" +) + +// errAlreadyRunning is returned when a start is requested while a run is already +// in flight. +var errAlreadyRunning = errors.New("client is already running") + +// lifecycleOp is a serialized lifecycle operation processed by the supervisor. +type lifecycleOp int + +const ( + opStart lifecycleOp = iota + opStop +) + +// lifecycleCmd is a single start/stop request handed to the supervisor goroutine. +// done is the caller-supplied notification channel (nil for fire-and-forget): +// - for opStart it receives the run's end result when the run terminates, or +// errAlreadyRunning immediately if a run is already in flight. +// - for opStop it receives nil once the in-flight run has fully unwound. +type lifecycleCmd struct { + op lifecycleOp + mobileDep MobileDependency + runningChan chan struct{} + logPath string + done chan error +} + +// runEndResult is sent by the run goroutine to the supervisor when a run ends, +// whether on its own (error / external context cancellation) or because of a Stop. +type runEndResult struct { + err error +} + +// runFunc executes a single client run bound to the supervisor-owned context. +type runFunc func(ctx context.Context, mobileDep MobileDependency, runningChan chan struct{}, logPath string) error + +// supervisor serializes start/stop of a single client run. Every request goes +// through cmdCh and is handled one at a time by the loop goroutine, so two +// lifecycle operations can never overlap and their order is preserved (FIFO). +// The loop goroutine is the sole owner of runCancel/runWaiter, so that state +// needs no locking. The loop exits when the parent context is cancelled. +type supervisor struct { + ctx context.Context + run runFunc + cmdCh chan lifecycleCmd + runEnded chan runEndResult + + // owned exclusively by the loop goroutine. A non-nil runCancel means a run + // is in flight; runWaiter is whoever asked to be notified when it ends. + runCancel context.CancelFunc + runWaiter chan error +} + +func newSupervisor(ctx context.Context, run runFunc) *supervisor { + s := &supervisor{ + ctx: ctx, + run: run, + cmdCh: make(chan lifecycleCmd, 16), + runEnded: make(chan runEndResult, 1), + } + go s.loop() + return s +} + +func (s *supervisor) loop() { + for { + select { + case <-s.ctx.Done(): + s.shutdown() + return + case cmd := <-s.cmdCh: + switch cmd.op { + case opStart: + s.handleStart(cmd) + case opStop: + s.handleStop(cmd) + } + case res := <-s.runEnded: + // Run ended on its own, without an explicit Stop. + s.finishRun(res.err) + } + } +} + +func (s *supervisor) handleStart(cmd lifecycleCmd) { + if s.runCancel != nil { + notify(cmd.done, errAlreadyRunning) + return + } + + runCtx, cancel := context.WithCancel(s.ctx) + s.runCancel = cancel + s.runWaiter = cmd.done + + go func(ctx context.Context, m MobileDependency, rc chan struct{}, lp string) { + err := s.run(ctx, m, rc, lp) + s.runEnded <- runEndResult{err: err} + }(runCtx, cmd.mobileDep, cmd.runningChan, cmd.logPath) +} + +func (s *supervisor) handleStop(cmd lifecycleCmd) { + if s.runCancel == nil { + notify(cmd.done, nil) + return + } + + // Cancel the in-flight run and block the supervisor until it has fully + // unwound, so the next queued command (e.g. a fresh start) starts from a + // clean slate. The run goroutine reports completion via runEnded. + s.runCancel() + res := <-s.runEnded + s.finishRun(res.err) + notify(cmd.done, nil) +} + +// finishRun resets lifecycle state after a run terminates and hands the run +// error back to whoever asked to be notified of the start. +func (s *supervisor) finishRun(err error) { + s.runCancel = nil + if s.runWaiter != nil { + notify(s.runWaiter, err) + s.runWaiter = nil + } +} + +// shutdown tears down the in-flight run when the parent context is cancelled, +// then fails any still-queued commands so their callers never hang. +func (s *supervisor) shutdown() { + if s.runCancel != nil { + s.runCancel() + res := <-s.runEnded + s.finishRun(res.err) + } + for { + select { + case cmd := <-s.cmdCh: + notify(cmd.done, s.ctx.Err()) + default: + return + } + } +} + +// 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(mobileDep MobileDependency, runningChan chan struct{}, logPath string, done chan error) { + cmd := lifecycleCmd{op: opStart, mobileDep: mobileDep, runningChan: runningChan, logPath: logPath, done: done} + select { + case s.cmdCh <- cmd: + case <-s.ctx.Done(): + notify(done, s.ctx.Err()) + } +} + +// start enqueues a start and blocks until the run terminates, preserving the +// blocking contract of the legacy Run entry points. +func (s *supervisor) start(mobileDep MobileDependency, runningChan chan struct{}, logPath string) error { + done := make(chan error, 1) + s.startAsync(mobileDep, runningChan, logPath, done) + select { + case err := <-done: + return err + case <-s.ctx.Done(): + return s.ctx.Err() + } +} + +// stop enqueues a stop and blocks until the in-flight run is fully torn down. +func (s *supervisor) stop() error { + done := make(chan error, 1) + select { + case s.cmdCh <- lifecycleCmd{op: opStop, done: done}: + case <-s.ctx.Done(): + return s.ctx.Err() + } + select { + case err := <-done: + return err + case <-s.ctx.Done(): + return s.ctx.Err() + } +} + +// notify sends on a caller-supplied channel without blocking. The channel is +// expected to be buffered (cap 1); a nil channel means the caller did not ask +// to be notified. +func notify(ch chan error, err error) { + if ch == nil { + return + } + select { + case ch <- err: + default: + } +}