diff --git a/client/embed/embed.go b/client/embed/embed.go index 0e8991be2..d0d88b177 100644 --- a/client/embed/embed.go +++ b/client/embed/embed.go @@ -279,9 +279,11 @@ func (c *Client) Start(startCtx context.Context) error { 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. + // ConnectClient.Stop now cancels its own run context and waits for the + // run loop to tear the engine down, so this cancel() is no longer + // required to break the deadlock and could be removed. It is kept as a + // defensive belt-and-suspenders: cancelling the parent context first + // guarantees the run loop is unblocked even if Stop's contract regresses. cancel() if stopErr := client.Stop(); stopErr != nil { return fmt.Errorf("stop error after context done. Stop error: %w. Context done: %w", stopErr, startCtx.Err()) diff --git a/client/internal/connect.go b/client/internal/connect.go index d93b62bb5..7cd2bab22 100644 --- a/client/internal/connect.go +++ b/client/internal/connect.go @@ -11,6 +11,7 @@ import ( "runtime/debug" "strings" "sync" + "sync/atomic" "time" "github.com/cenkalti/backoff/v4" @@ -54,6 +55,10 @@ var androidRunOverride func(c *ConnectClient, runningChan chan struct{}, logPath type ConnectClient struct { ctx context.Context + runCancel context.CancelFunc + runExited chan struct{} + runOnce sync.Once + runStarted atomic.Bool config *profilemanager.Config statusRecorder *peer.Status @@ -70,8 +75,14 @@ func NewConnectClient( config *profilemanager.Config, statusRecorder *peer.Status, ) *ConnectClient { + // Derive the run context here so Stop owns the cancel that unblocks the run + // loop. runCancel is set once at construction, so Stop can call it without + // racing the run loop's startup. Callers therefore need not cancel before Stop. + runCtx, runCancel := context.WithCancel(ctx) return &ConnectClient{ - ctx: ctx, + ctx: runCtx, + runCancel: runCancel, + runExited: make(chan struct{}), config: config, statusRecorder: statusRecorder, engineMutex: sync.Mutex{}, @@ -135,6 +146,11 @@ func (c *ConnectClient) RunOniOS( } func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan struct{}, logPath string) error { + // Mark the loop as started and signal exit on return so Stop can wait for + // the loop to finish (and skip the wait if the loop never ran). + c.runStarted.Store(true) + defer c.runOnce.Do(func() { close(c.runExited) }) + defer func() { if r := recover(); r != nil { rec := c.statusRecorder @@ -290,7 +306,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan log.Debug(err) if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) { state.Set(StatusNeedsLogin) - _ = c.Stop() + c.runCancel() return backoff.Permanent(wrapErr(err)) // unrecoverable error } return wrapErr(err) @@ -410,14 +426,10 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan c.engine = nil c.engineMutex.Unlock() - // todo: consider to remove this condition. Is not thread safe. - // We should always call Stop(), but we need to verify that it is idempotent - if engine.wgInterface != nil { - log.Infof("ensuring %s is removed, Netbird engine context cancelled", engine.wgInterface.Name()) + log.Infof("ensuring wg interface is removed, Netbird engine context cancelled") - if err := engine.Stop(); err != nil { - log.Errorf("Failed to stop engine: %v", err) - } + if err := engine.Stop(); err != nil { + log.Errorf("Failed to stop engine: %v", err) } c.statusRecorder.ClientTeardown() @@ -433,12 +445,12 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan } c.statusRecorder.ClientStart() - err = backoff.Retry(operation, backOff) + err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx)) if err != nil { 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() + c.runCancel() } return err } @@ -516,11 +528,9 @@ func (c *ConnectClient) Status() StatusType { } func (c *ConnectClient) Stop() error { - engine := c.Engine() - if engine != nil { - if err := engine.Stop(); err != nil { - return fmt.Errorf("stop engine: %w", err) - } + c.runCancel() + if c.runStarted.Load() { + <-c.runExited } return nil } diff --git a/client/internal/engine.go b/client/internal/engine.go index 42712da92..452075da8 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -86,6 +86,8 @@ const ( var ErrResetConnection = fmt.Errorf("reset connection") +var ErrEngineAlreadyStarted = errors.New("engine already started") + type EngineConfig struct { WgPort int WgIfaceName string @@ -199,6 +201,8 @@ type Engine struct { ctx context.Context cancel context.CancelFunc + started bool + wgInterface WGIface udpMux *udpmux.UniversalUDPMuxDefault @@ -279,9 +283,15 @@ func NewEngine( services EngineServices, mobileDep MobileDependency, ) *Engine { + // The engine is single-use: a fresh instance is built per connection + // cycle (see Client.run), so the run context is created once here rather + // than in Start. + ctx, cancel := context.WithCancel(clientCtx) engine := &Engine{ clientCtx: clientCtx, clientCancel: clientCancel, + ctx: ctx, + cancel: cancel, signal: services.SignalClient, signaler: peer.NewSignaler(services.SignalClient, config.WgPrivateKey), mgmClient: services.MgmClient, @@ -314,8 +324,34 @@ func (e *Engine) Stop() error { log.Debugf("tried stopping engine that is nil") return nil } + e.cancel() e.syncMsgMux.Lock() + e.stopLocked() + + e.syncMsgMux.Unlock() + + timeout := e.calculateShutdownTimeout() + log.Debugf("waiting for goroutines to finish with timeout: %v", timeout) + shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + if err := waitWithContext(shutdownCtx, &e.shutdownWg); err != nil { + log.Warnf("shutdown timeout exceeded after %v, some goroutines may still be running", timeout) + } + + log.Infof("stopped Netbird Engine") + + return nil +} + +// stopLocked tears down everything Start may have brought up, in the order +// teardown requires (DNS before the interface goes down, flow manager after). +// The caller must hold syncMsgMux. It is shared by Stop and by Start's failure +// path, so a partially-initialized engine is cleaned up the same way; every +// step is nil-guarded. It does not wait on shutdownWg — the caller does that +// after releasing the lock, since the goroutines also take syncMsgMux. +func (e *Engine) stopLocked() { if e.connMgr != nil { e.connMgr.Close() } @@ -366,10 +402,6 @@ func (e *Engine) Stop() error { // so dbus and friends don't complain because of a missing interface e.stopDNSServer() - if e.cancel != nil { - e.cancel() - } - e.jobExecutorWG.Wait() // block until job goroutines finish e.close() @@ -388,21 +420,6 @@ func (e *Engine) Stop() error { if err := e.stateManager.PersistState(context.Background()); err != nil { log.Errorf("failed to persist state: %v", err) } - - e.syncMsgMux.Unlock() - - timeout := e.calculateShutdownTimeout() - log.Debugf("waiting for goroutines to finish with timeout: %v", timeout) - shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - if err := waitWithContext(shutdownCtx, &e.shutdownWg); err != nil { - log.Warnf("shutdown timeout exceeded after %v, some goroutines may still be running", timeout) - } - - log.Infof("stopped Netbird Engine") - - return nil } // calculateShutdownTimeout returns shutdown timeout: 10s base + 100ms per peer, capped at 30s. @@ -440,18 +457,38 @@ func waitWithContext(ctx context.Context, wg *sync.WaitGroup) error { // Start creates a new WireGuard tunnel interface and listens to events from Signal and Management services // Connections to remote peers are not established here. // However, they will be established once an event with a list of peers to connect to will be received from Management Service -func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) error { +func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) (err error) { e.syncMsgMux.Lock() defer e.syncMsgMux.Unlock() - if err := iface.ValidateMTU(e.config.MTU); err != nil { + // The engine is single-use. Reject a duplicate start and a start on an + // already-stopped engine (run context cancelled). + if e.started { + return ErrEngineAlreadyStarted + } + + if ctxErr := e.ctx.Err(); ctxErr != nil { + return fmt.Errorf("engine already stopped: %w", ctxErr) + } + + e.started = true + + // Tear down any partially-initialized state on a failed start. Cancel the + // run context first so goroutines started before the failure (connMgr, + // srWatcher, monitors) unwind, then stopLocked mirrors Stop's teardown (we + // already hold syncMsgMux), cleaning up route/DNS/flow/state managers too, + // not just what close() covers. + defer func() { + if err != nil { + e.cancel() + e.stopLocked() + } + }() + + if err = iface.ValidateMTU(e.config.MTU); err != nil { return fmt.Errorf("invalid MTU configuration: %w", err) } - if e.cancel != nil { - e.cancel() - } - e.ctx, e.cancel = context.WithCancel(e.clientCtx) e.exposeManager = expose.NewManager(e.ctx, e.mgmClient) wgIface, err := e.newWgIface() @@ -485,13 +522,11 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) initialRoutes, dnsConfig, dnsFeatureFlag, err := e.readInitialSettings() if err != nil { - e.close() return fmt.Errorf("read initial settings: %w", err) } dnsServer, err := e.newDnsServer(dnsConfig) if err != nil { - e.close() return fmt.Errorf("create dns server: %w", err) } e.dnsServer = dnsServer @@ -526,7 +561,6 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) if err = e.wgInterfaceCreate(); err != nil { log.Errorf("failed creating tunnel interface %s: [%s]", e.config.WgIfaceName, err.Error()) - e.close() return fmt.Errorf("create wg interface: %w", err) } @@ -535,7 +569,6 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) } if err := e.createFirewall(); err != nil { - e.close() return err } @@ -547,7 +580,6 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) e.udpMux, err = e.wgInterface.Up() if err != nil { log.Errorf("failed to pull up wgInterface [%s]: %s", e.wgInterface.Name(), err.Error()) - e.close() return fmt.Errorf("up wg interface: %w", err) } @@ -572,9 +604,7 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) e.acl = acl.NewDefaultManager(e.firewall) } - err = e.dnsServer.Initialize() - if err != nil { - e.close() + if err := e.dnsServer.Initialize(); err != nil { return fmt.Errorf("initialize dns server: %w", err) } @@ -586,7 +616,9 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) e.srWatcher = guard.NewSRWatcher(e.signal, e.relayManager, e.mobileDep.IFaceDiscover, iceCfg) e.srWatcher.Start(peer.IsForceRelayed()) - e.receiveSignalEvents() + if err = e.receiveSignalEvents(); err != nil { + return err + } e.receiveManagementEvents() e.receiveJobEvents() @@ -638,7 +670,6 @@ func (e *Engine) createFirewall() error { func (e *Engine) initFirewall() error { if err := e.routeManager.SetFirewall(e.firewall); err != nil { - e.close() return fmt.Errorf("set firewall: %w", err) } @@ -1698,7 +1729,7 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentV } // receiveSignalEvents connects to the Signal Service event stream to negotiate connection with remote peers -func (e *Engine) receiveSignalEvents() { +func (e *Engine) receiveSignalEvents() error { e.shutdownWg.Add(1) go func() { defer e.shutdownWg.Done() @@ -1769,7 +1800,12 @@ func (e *Engine) receiveSignalEvents() { } }() - e.signal.WaitStreamConnected() + // todo: consider to remove this blocker. I do not see benefit to block the Start operations + e.signal.WaitStreamConnected(e.ctx) + if err := e.ctx.Err(); err != nil { + return fmt.Errorf("wait for signal stream: %w", err) + } + return nil } func (e *Engine) parseNATExternalIPMappings() []string { diff --git a/client/internal/engine_test.go b/client/internal/engine_test.go index 289f1906f..8f29bf072 100644 --- a/client/internal/engine_test.go +++ b/client/internal/engine_test.go @@ -247,7 +247,7 @@ func TestEngine_SSH(t *testing.T) { return } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(CtxInitState(context.Background())) defer cancel() relayMgr := relayClient.NewManager(ctx, nil, key.PublicKey().String(), iface.DefaultMTU) @@ -426,7 +426,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) { return } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(CtxInitState(context.Background())) defer cancel() relayMgr := relayClient.NewManager(ctx, nil, key.PublicKey().String(), iface.DefaultMTU) @@ -638,7 +638,7 @@ func TestEngine_Sync(t *testing.T) { return } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(CtxInitState(context.Background())) defer cancel() // feed updates to Engine via mocked Management client @@ -817,7 +817,7 @@ func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) { return } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(CtxInitState(context.Background())) defer cancel() wgIfaceName := fmt.Sprintf("utun%d", 104+n) @@ -1024,7 +1024,7 @@ func TestEngine_UpdateNetworkMapWithDNSUpdate(t *testing.T) { return } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(CtxInitState(context.Background())) defer cancel() wgIfaceName := fmt.Sprintf("utun%d", 104+n) diff --git a/client/server/server.go b/client/server/server.go index a4d53a823..3f6dabc56 100644 --- a/client/server/server.go +++ b/client/server/server.go @@ -993,6 +993,10 @@ func (s *Server) cleanupConnection() error { return nil } + // TODO: consider calling s.connectClient.Stop() instead of engine.Stop(). + // actCancel() lets the run loop stop the engine too, so both stop it + // concurrently; ConnectClient.Stop cancels and waits for the run loop, + // making the run loop the sole owner of engine shutdown. if engine != nil { if err := engine.Stop(); err != nil { return err diff --git a/shared/signal/client/client.go b/shared/signal/client/client.go index 9dc6ccd37..fb77cb90f 100644 --- a/shared/signal/client/client.go +++ b/shared/signal/client/client.go @@ -33,7 +33,7 @@ type Client interface { Receive(ctx context.Context, msgHandler func(msg *proto.Message) error) error Ready() bool IsHealthy() bool - WaitStreamConnected() + WaitStreamConnected(context.Context) SendToStream(msg *proto.EncryptedMessage) error Send(msg *proto.Message) error SetOnReconnectedListener(func()) diff --git a/shared/signal/client/client_test.go b/shared/signal/client/client_test.go index 1af34e37a..41def08a1 100644 --- a/shared/signal/client/client_test.go +++ b/shared/signal/client/client_test.go @@ -65,7 +65,10 @@ var _ = Describe("GrpcClient", func() { return } }() - clientA.WaitStreamConnected() + ctxA, cancelA := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelA() + clientA.WaitStreamConnected(ctxA) + Expect(clientA.StreamConnected()).To(BeTrue()) // connect PeerB to Signal keyB, _ := wgtypes.GenerateKey() @@ -91,7 +94,10 @@ var _ = Describe("GrpcClient", func() { } }() - clientB.WaitStreamConnected() + ctxB, cancelB := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelB() + clientB.WaitStreamConnected(ctxB) + Expect(clientB.StreamConnected()).To(BeTrue()) // PeerA initiates ping-pong err := clientA.Send(&sigProto.Message{ @@ -129,8 +135,10 @@ var _ = Describe("GrpcClient", func() { return } }() - client.WaitStreamConnected() - Expect(client).NotTo(BeNil()) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + client.WaitStreamConnected(ctx) + Expect(client.StreamConnected()).To(BeTrue()) }) }) diff --git a/shared/signal/client/grpc.go b/shared/signal/client/grpc.go index eb18cea05..2086e0fe6 100644 --- a/shared/signal/client/grpc.go +++ b/shared/signal/client/grpc.go @@ -246,15 +246,6 @@ func (c *GrpcClient) notifyStreamConnected() { } } -func (c *GrpcClient) getStreamStatusChan() <-chan struct{} { - c.mux.Lock() - defer c.mux.Unlock() - if c.connectedCh == nil { - c.connectedCh = make(chan struct{}) - } - return c.connectedCh -} - func (c *GrpcClient) connect(ctx context.Context, key string) (proto.SignalExchange_ConnectStreamClient, error) { c.stream = nil @@ -310,14 +301,24 @@ func (c *GrpcClient) IsHealthy() bool { } // WaitStreamConnected waits until the client is connected to the Signal stream -func (c *GrpcClient) WaitStreamConnected() { - +func (c *GrpcClient) WaitStreamConnected(ctx context.Context) { + // Check the status and obtain the wait channel atomically: otherwise + // notifyStreamConnected could flip the status and close/clear the channel + // between the check and the channel creation, leaving us waiting forever on + // a stale channel. + c.mux.Lock() if c.status == StreamConnected { + c.mux.Unlock() return } + if c.connectedCh == nil { + c.connectedCh = make(chan struct{}) + } + ch := c.connectedCh + c.mux.Unlock() - ch := c.getStreamStatusChan() select { + case <-ctx.Done(): case <-c.ctx.Done(): case <-ch: } diff --git a/shared/signal/client/mock.go b/shared/signal/client/mock.go index 95381a5b0..0c8a083c5 100644 --- a/shared/signal/client/mock.go +++ b/shared/signal/client/mock.go @@ -55,7 +55,7 @@ func (sm *MockClient) Ready() bool { return sm.ReadyFunc() } -func (sm *MockClient) WaitStreamConnected() { +func (sm *MockClient) WaitStreamConnected(context.Context) { if sm.WaitStreamConnectedFunc == nil { return }