Clarifies: service -> ServiceRunning -> up -> ConnectionRunning -> connestablished ->connEstablished -> end of run -> connDone

This commit is contained in:
riccardom
2026-06-18 14:40:03 +02:00
parent 6465997a69
commit ff98105212
2 changed files with 22 additions and 20 deletions
+3 -3
View File
@@ -142,12 +142,12 @@ func (s *Server) restartEngineForMDMLocked() error {
_, cancel := context.WithCancel(s.rootCtx) _, cancel := context.WithCancel(s.rootCtx)
s.actCancel = cancel s.actCancel = cancel
s.clientRunningChan = make(chan struct{}) s.connectionEstablishedChan = make(chan struct{})
s.clientDoneChan = make(chan error, 1) s.connectionDoneChan = make(chan error, 1)
log.Info("MDM restart: starting a fresh run with re-resolved config") log.Info("MDM restart: starting a fresh run with re-resolved config")
// MDM restart has no incoming RPC metadata; fire and forget (the supervisor // MDM restart has no incoming RPC metadata; fire and forget (the supervisor
// reconnects internally and we don't block on the run). // reconnects internally and we don't block on the run).
s.connectClient.RunAsync(config, nil, s.clientRunningChan, s.clientDoneChan) s.connectClient.RunAsync(config, nil, s.connectionEstablishedChan, s.connectionDoneChan)
s.publishConfigChangedEvent("mdm") s.publishConfigChangedEvent("mdm")
return nil return nil
} }
+19 -17
View File
@@ -64,8 +64,8 @@ type Server struct {
proto.UnimplementedDaemonServiceServer proto.UnimplementedDaemonServiceServer
// Whether a run is in flight is owned by the supervisor // Whether a run is in flight is owned by the supervisor
// (connectClient.ConnectionRunning); the daemon keeps no separate flag. // (connectClient.ConnectionRunning); the daemon keeps no separate flag.
clientRunningChan chan struct{} // closed by the run when the engine is ready connectionEstablishedChan chan struct{} // closed by the run once the connection is established (StatusConnected)
clientDoneChan chan error // receives the run's end result connectionDoneChan chan error // receives the run's end result
connectClient *internal.ConnectClient connectClient *internal.ConnectClient
@@ -227,11 +227,11 @@ func (s *Server) Start() error {
return nil return nil
} }
s.clientRunningChan = make(chan struct{}) s.connectionEstablishedChan = make(chan struct{})
s.clientDoneChan = make(chan error, 1) s.connectionDoneChan = make(chan error, 1)
// Boot autoconnect: no incoming RPC metadata. The supervisor runs the // Boot autoconnect: no incoming RPC metadata. The supervisor runs the
// client and reconnects internally; we just fire and forget. // client and reconnects internally; we just fire and forget.
s.connectClient.RunAsync(config, nil, s.clientRunningChan, s.clientDoneChan) s.connectClient.RunAsync(config, nil, s.connectionEstablishedChan, s.connectionDoneChan)
s.publishConfigChangedEvent("startup") s.publishConfigChangedEvent("startup")
return nil return nil
} }
@@ -753,10 +753,10 @@ func (s *Server) Up(callerCtx context.Context, msg *proto.UpRequest) (*proto.UpR
s.statusRecorder.UpdateManagementAddress(s.config.ManagementURL.String()) s.statusRecorder.UpdateManagementAddress(s.config.ManagementURL.String())
s.statusRecorder.UpdateRosenpass(s.config.RosenpassEnabled, s.config.RosenpassPermissive) s.statusRecorder.UpdateRosenpass(s.config.RosenpassEnabled, s.config.RosenpassPermissive)
s.clientRunningChan = make(chan struct{}) s.connectionEstablishedChan = make(chan struct{})
s.clientDoneChan = make(chan error, 1) s.connectionDoneChan = make(chan error, 1)
s.connectClient.RunAsync(s.config, md, s.clientRunningChan, s.clientDoneChan) s.connectClient.RunAsync(s.config, md, s.connectionEstablishedChan, s.connectionDoneChan)
s.publishConfigChangedEvent("up_rpc") s.publishConfigChangedEvent("up_rpc")
s.mutex.Unlock() s.mutex.Unlock()
@@ -767,11 +767,14 @@ func (s *Server) waitForUp(callerCtx context.Context) (*proto.UpResponse, error)
timeoutCtx, cancel := context.WithTimeout(callerCtx, 50*time.Second) timeoutCtx, cancel := context.WithTimeout(callerCtx, 50*time.Second)
defer cancel() defer cancel()
// Snapshot the per-run channels under the lock so a concurrent Up that // Read the per-run channels under the lock. They are written under s.mutex
// replaces them cannot race the select below. // by Up/Start and by the MDM restart (restartEngineForMDMLocked, which runs
// on the ticker goroutine), so reading them here — where the Up caller has
// already released the lock — must be synchronized both to avoid a data race
// and to capture a consistent (established, done) pair from the same run.
s.mutex.Lock() s.mutex.Lock()
runningChan := s.clientRunningChan establishedChan := s.connectionEstablishedChan
doneChan := s.clientDoneChan doneChan := s.connectionDoneChan
s.mutex.Unlock() s.mutex.Unlock()
select { select {
@@ -781,7 +784,7 @@ func (s *Server) waitForUp(callerCtx context.Context) (*proto.UpResponse, error)
return nil, fmt.Errorf("client failed to connect: %w", err) return nil, fmt.Errorf("client failed to connect: %w", err)
} }
return nil, fmt.Errorf("client stopped before becoming ready") return nil, fmt.Errorf("client stopped before becoming ready")
case <-runningChan: case <-establishedChan:
s.isSessionActive.Store(true) s.isSessionActive.Store(true)
return &proto.UpResponse{}, nil return &proto.UpResponse{}, nil
case <-callerCtx.Done(): case <-callerCtx.Done():
@@ -1080,14 +1083,14 @@ func (s *Server) Status(
// poll-and-cancel: we just wait for the run to become ready or to end. // poll-and-cancel: we just wait for the run to become ready or to end.
s.mutex.Lock() s.mutex.Lock()
client := s.connectClient client := s.connectClient
runningChan := s.clientRunningChan establishedChan := s.connectionEstablishedChan
doneChan := s.clientDoneChan doneChan := s.connectionDoneChan
s.mutex.Unlock() s.mutex.Unlock()
alive := client.ConnectionRunning() alive := client.ConnectionRunning()
if msg.WaitForReady != nil && *msg.WaitForReady && alive { if msg.WaitForReady != nil && *msg.WaitForReady && alive {
select { select {
case <-runningChan: case <-establishedChan:
case <-doneChan: case <-doneChan:
case <-ctx.Done(): case <-ctx.Done():
return nil, ctx.Err() return nil, ctx.Err()
@@ -1678,7 +1681,6 @@ func (s *Server) onSessionExpire() {
} }
} }
// sendTerminalNotification sends a terminal notification message // sendTerminalNotification sends a terminal notification message
// to inform the user that the NetBird connection session has expired. // to inform the user that the NetBird connection session has expired.
func sendTerminalNotification() error { func sendTerminalNotification() error {