fix(cli): port runtime lifecycle cleanup

cmd/cli/dns_proxy.go

@@ -207,6 +207,7 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha
 				defer s.Shutdown()
 				select {
 				case <-p.stopCh:
+				case <-p.runAbortCh:
 				case <-gctx.Done():
 				case err := <-errCh:
 					p.Warn().Err(err).Msg("Local IPv6 listener failed")
@@ -227,6 +228,7 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha
 						defer s.Shutdown()
 						select {
 						case <-p.stopCh:
+						case <-p.runAbortCh:
 						case <-gctx.Done():
 						case err := <-errCh:
 							p.Warn().Err(err).Msgf("Could not listen on %s: %s", proto, listenAddr)
@@ -242,9 +244,18 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha
 			addr := net.JoinHostPort(cfg.IP, strconv.Itoa(cfg.Port))
 			s, errCh := runDNSServer(addr, proto, handler)
 			defer s.Shutdown()
-			p.started <- struct{}{}
+			select {
+			case p.started <- struct{}{}:
+			case <-p.stopCh:
+				return nil
+			case <-p.runAbortCh:
+				return nil
+			case <-gctx.Done():
+				return nil
+			}
 			select {
 			case <-p.stopCh:
+			case <-p.runAbortCh:
 			case <-gctx.Done():
 			case err := <-errCh:
 				return err
@@ -1479,7 +1490,7 @@ func runDNSServer(addr, network string, handler dns.Handler) (*dns.Server, <-cha
 	startedCh := make(chan struct{})
 	s.NotifyStartedFunc = func() { sync.OnceFunc(func() { close(startedCh) })() }
 
-	errCh := make(chan error)
+	errCh := make(chan error, 1)
 	go func() {
 		defer close(errCh)
 		if err := s.ListenAndServe(); err != nil {
@@ -1642,6 +1653,10 @@ func (p *prog) selfUninstallCoolOfPeriod() {
 // forceFetchingAPI sends signal to force syncing API config if run in cd mode,
 // and the domain == "cdUID.verify.controld.com"
 func (p *prog) forceFetchingAPI(domain string) {
+	if !p.beginNetworkActivity() {
+		return
+	}
+	defer p.netMonitorWG.Done()
 	if cdUID == "" {
 		return
 	}
@@ -1658,9 +1673,32 @@ func (p *prog) forceFetchingAPI(domain string) {
 		return
 	}
 	_ = p.apiForceReloadGroup.DoChan("force_sync_api", func() (interface{}, error) {
-		p.apiForceReloadCh <- struct{}{}
+		if !p.beginNetworkActivity() {
+			return nil, nil
+		}
+		defer p.netMonitorWG.Done()
+		shutdown := p.networkActivityDone()
+		select {
+		case <-shutdown:
+			return nil, nil
+		case p.apiForceReloadCh <- struct{}{}:
+		case <-p.stopCh:
+			return nil, nil
+		case <-p.runAbortCh:
+			return nil, nil
+		}
 		// Wait here to prevent abusing API if we are flooded.
-		time.Sleep(timeDurationOrDefault(p.cfg.Service.ForceRefetchWaitTime, 30) * time.Second)
+		p.mu.Lock()
+		wait := timeDurationOrDefault(p.cfg.Service.ForceRefetchWaitTime, 30) * time.Second
+		p.mu.Unlock()
+		timer := time.NewTimer(wait)
+		defer timer.Stop()
+		select {
+		case <-timer.C:
+		case <-shutdown:
+		case <-p.stopCh:
+		case <-p.runAbortCh:
+		}
 		return nil, nil
 	})
 }
@@ -1861,6 +1899,18 @@ func FlushDNSCache() error {
 	return nil
 }
 
+// networkChangeCallback fences even callbacks already dispatched by netmon
+// when Close begins. The monitor does not track those callback goroutines.
+func (p *prog) networkChangeCallback(fn netmon.ChangeFunc) netmon.ChangeFunc {
+	return func(delta *netmon.ChangeDelta) {
+		if !p.beginNetworkActivity() {
+			return
+		}
+		defer p.netMonitorWG.Done()
+		fn(delta)
+	}
+}
+
 // monitorNetworkChanges starts monitoring for network interface changes
 func (p *prog) monitorNetworkChanges(ctx context.Context) error {
 	mon, err := newNetworkChangeMonitorFn(func(format string, args ...any) {
@@ -1871,9 +1921,14 @@ func (p *prog) monitorNetworkChanges(ctx context.Context) error {
 		return fmt.Errorf("creating network monitor: %w", err)
 	}
 
-	mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) {
+	mon.RegisterChangeCallback(p.networkChangeCallback(func(delta *netmon.ChangeDelta) {
 		p.handleNetworkChange(ctx, delta, mon.IsMajorChangeFrom(delta.Old, delta.New))
-	})
+	}))
+	if !p.setNetMonitor(mon) {
+		_ = mon.Close()
+		p.Debug().Msg("Network monitor discarded, ctrld is stopping")
+		return nil
+	}
 
 	mon.Start()
 	p.logger.Load().Debug().Msg("Network monitor started")
@@ -2421,12 +2476,19 @@ func (p *prog) debounceRecovery(transitionID uint64) {
 	}
 	p.recoveryDebounceMu.Lock()
 	defer p.recoveryDebounceMu.Unlock()
+	if p.networkActivityClosed() {
+		return
+	}
 
 	if p.recoveryDebounceTimer != nil {
 		p.recoveryDebounceTimer.Stop()
 		p.Debug().Msg("Recovery debounce: resetting timer (rapid network change)")
 	}
 	p.recoveryDebounceTimer = time.AfterFunc(recoveryDebounceWindow, func() {
+		if !p.beginNetworkActivity() {
+			return
+		}
+		defer p.netMonitorWG.Done()
 		p.networkSourceMu.Lock()
 		if p.networkAcceptedGen.Load() != transitionID {
 			p.networkSourceMu.Unlock()
@@ -2488,6 +2550,10 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 }
 
 func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID uint64) {
+	if !p.beginNetworkActivity() {
+		return
+	}
+	defer p.netMonitorWG.Done()
 	p.networkSourceMu.Lock()
 	if transitionID != 0 && p.networkAcceptedGen.Load() != transitionID {
 		p.networkSourceMu.Unlock()
@@ -2508,6 +2574,14 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u
 	targetBefore := p.interceptTargetSnapshot()
 	outcome := recoveryOutcomeSuperseded
 	defer func() { p.endRecovery(&diagnostic, outcome, targetBefore) }()
+	// Clean every exit before the final snapshot, including cancellation before mutation.
+	defer p.recoveryCanceledCleanup(gen)
+	if recoveryCtx.Err() != nil {
+		if p.recoveryGen.Load() == gen {
+			outcome = recoveryOutcomeCanceled
+		}
+		return
+	}
 	if reason == RecoveryReasonNetworkChange {
 		p.logger.Load().Debug().Msg("Network change recovery now owns shared recovery state")
 	}
@@ -2539,10 +2613,9 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u
 	upstreams := p.buildRecoveryUpstreams(reason)
 	recovered, err := waitForUpstreamRecoveryFn(p, recoveryCtx, upstreams, &diagnostic)
 	if err != nil {
-		if p.recoveryOwnsState(gen) {
+		if p.recoveryGen.Load() == gen {
 			outcome = recoveryOutcomeCanceled
 		}
-		p.recoveryCanceledCleanup(gen)
 		return
 	}
 	diagnostic.recoveredUpstream = recovered
@@ -2702,6 +2775,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 	var wg sync.WaitGroup
 
 	p.Debug().Msgf("Starting upstream recovery check for %d upstreams", len(upstreams))
+	defer func() { cancel(); wg.Wait() }()
 
 	for name, uc := range upstreams {
 		wg.Add(1)
@@ -2781,7 +2855,6 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 	case <-ctx.Done():
 		return "", ctx.Err()
 	}
-	wg.Wait()
 	return recovered, nil
 }