fix(cli): release runtime resources across startup and shutdown

Co-authored-by: Ginder Singh <ginder@windscribe.com>

cmd/cli/api_reload_shutdown_test.go

@@ -0,0 +1,157 @@
+package cli
+
+import (
+	"context"
+	"encoding/base64"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/Control-D-Inc/ctrld"
+	"github.com/Control-D-Inc/ctrld/internal/controld"
+)
+
+func TestAPIConfigReloadShutdown(t *testing.T) {
+	const env = "CTRLD_TEST_API_RELOAD_SHUTDOWN"
+	mode := os.Getenv(env)
+	if mode == "" {
+		for _, mode := range []string{"stop-exclude", "abort-exclude", "stop-custom", "abort-custom", "stop-internal", "abort-internal", "stop-fetch", "abort-fetch", "live-exclude", "live-custom", "live-internal"} {
+			t.Run(mode, func(t *testing.T) {
+				ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+				defer cancel()
+				cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestAPIConfigReloadShutdown$", "-test.v")
+				cmd.Env = append(os.Environ(), env+"="+mode)
+				out, err := cmd.CombinedOutput()
+				require.NoError(t, err, "%s", out)
+			})
+		}
+		return
+	}
+	cdUID = "testuid"
+	version = "dev"
+	p := &prog{
+		cfg: &ctrld.Config{}, rc: &controld.ResolverConfig{},
+		stopCh: make(chan struct{}), runAbortCh: make(chan struct{}),
+		apiForceReloadCh: make(chan struct{}), apiReloadCh: make(chan *ctrld.Config),
+	}
+	rc := &controld.ResolverConfig{Exclude: []string{"internal.test"}}
+	if strings.HasSuffix(mode, "internal") {
+		rc.Exclude = nil
+		rc.SplitDNS = []controld.SplitDNS{{Domain: "internal.test", Mode: controld.SplitDNSModeOS}}
+	}
+	if strings.HasSuffix(mode, "custom") {
+		rc.Ctrld.CustomConfig = base64.StdEncoding.EncodeToString([]byte(`[listener.0]
+ip = "127.0.0.1"
+port = 5354
+[network.0]
+cidrs = ["127.0.0.0/8"]
+[upstream.0]
+type = "legacy"
+endpoint = "127.0.0.1"
+`))
+	}
+	entered, canceled := make(chan struct{}), make(chan struct{})
+	fetchResolverConfigForReload = func(ctx context.Context, _ *controld.ResolverConfigRequest, _ bool) (*controld.ResolverConfig, error) {
+		close(entered)
+		if strings.HasSuffix(mode, "fetch") {
+			<-ctx.Done()
+			close(canceled)
+			// A late successful response must not schedule a reload after stop.
+			return rc, nil
+		}
+		return rc, nil
+	}
+	done := make(chan struct{})
+	go func() { p.apiConfigReload(); close(done) }()
+	p.apiForceReloadCh <- struct{}{}
+	awaitShutdown(t, entered)
+	if strings.HasPrefix(mode, "live-") {
+		select {
+		case got := <-p.apiReloadCh:
+			if strings.HasSuffix(mode, "custom") {
+				require.NotNil(t, got)
+				require.Equal(t, "127.0.0.1", got.Listener["0"].IP)
+			} else {
+				require.Nil(t, got)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatal("running API loop did not deliver its reload")
+		}
+	} else if !strings.HasSuffix(mode, "fetch") {
+		marker := "exclude list changes detected, reloading"
+		if strings.HasSuffix(mode, "custom") {
+			marker = "custom config changes detected, reloading"
+		} else if strings.HasSuffix(mode, "internal") {
+			marker = "internal domain changes detected, reloading"
+		}
+		require.Eventually(t, func() bool { return strings.Contains(logOutput.String(), marker) }, time.Second, time.Millisecond)
+	}
+	if strings.HasPrefix(mode, "abort-") {
+		close(p.runAbortCh)
+	} else {
+		close(p.stopCh)
+	}
+	awaitShutdown(t, done)
+	if strings.HasSuffix(mode, "fetch") {
+		awaitShutdown(t, canceled)
+		require.Empty(t, p.rc.Exclude, "late fetch must not update run state")
+	}
+}
+
+func TestForceFetchingAPIStopsWithoutReceiver(t *testing.T) {
+	old := cdUID
+	cdUID = "testuid"
+	t.Cleanup(func() { cdUID = old })
+	for _, abort := range []bool{false, true} {
+		for _, delivered := range []bool{false, true} {
+			p := &prog{cfg: &ctrld.Config{}, stopCh: make(chan struct{}), runAbortCh: make(chan struct{}), apiForceReloadCh: make(chan struct{})}
+			p.forceFetchingAPI("testuid.verify.controld.com")
+			done := p.apiForceReloadGroup.DoChan("force_sync_api", func() (interface{}, error) { t.Error("lost active force-reload worker"); return nil, nil })
+			if delivered {
+				select {
+				case <-p.apiForceReloadCh:
+				case <-time.After(time.Second):
+					t.Fatal("no force-reload signal")
+				}
+			}
+			if abort {
+				close(p.runAbortCh)
+			} else {
+				close(p.stopCh)
+			}
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Fatal("force-reload send/cooldown survived stop")
+			}
+		}
+	}
+}
+
+type closeCountingLogConn struct {
+	net.Conn
+	closes atomic.Int32
+}
+
+func (c *closeCountingLogConn) Close() error              { c.closes.Add(1); return nil }
+func (*closeCountingLogConn) Write(b []byte) (int, error) { return len(b), nil }
+
+func TestStartupLogConnectionCloseOwnership(t *testing.T) {
+	c := &closeCountingLogConn{}
+	p := &prog{cfg: &ctrld.Config{}, dnsWatcherStopCh: make(chan struct{}), logConn: c}
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() { defer wg.Done(); p.notifyExitToLogServer(); p.closeLogConn() }()
+	}
+	wg.Wait()
+	require.NoError(t, p.shutdown())
+	require.EqualValues(t, 1, c.closes.Load())
+	require.Nil(t, p.logConn)
+}

cmd/cli/cli.go

@@ -50,6 +50,7 @@ const (
 	windowsForwardersFilename = ".forwarders.txt"
 	oldBinSuffix              = "_previous"
 	msgExit                   = "$$EXIT$$"
+	shutdownTimeout           = 10 * time.Second
 )
 
 var (
@@ -144,10 +145,15 @@ func initCLI() {
 }
 
 // isMobile reports whether the current OS is a mobile platform.
-func isMobile() bool {
+var isMobile = func() bool {
 	return runtime.GOOS == "android" || runtime.GOOS == "ios"
 }
 
+var (
+	networkUp = ctrldnet.Up
+	newRouter = router.New
+)
+
 func updateConfigInterceptMode(cfg *ctrld.Config, mode string) bool {
 	desired := ""
 	switch mode {
@@ -241,6 +247,8 @@ func run(appCallback *AppCallback, stopCh chan struct{}) {
 		dnsWatcherStopCh: make(chan struct{}),
 		apiReloadCh:      make(chan *ctrld.Config),
 		apiForceReloadCh: make(chan struct{}),
+		runDone:          make(chan struct{}),
+		runAbortCh:       make(chan struct{}),
 		cfg:              &cfg,
 		appCallback:      appCallback,
 	}
@@ -273,7 +281,15 @@ func run(appCallback *AppCallback, stopCh chan struct{}) {
 		return
 	}
 
-	if !daemon {
+	switch {
+	case isMobile():
+		defer p.finishRun()
+		// There is no OS service manager here, and s.Run parks a goroutine on a
+		// signal that never arrives, leaking one per start/stop cycle.
+		if err := p.Start(nil); err != nil {
+			mainLog.Load().Fatal().Err(err).Msg("failed to start ctrld")
+		}
+	case !daemon:
 		// We need to call s.Run() as soon as possible to response to the OS manager, so it
 		// can see ctrld is running and don't mark ctrld as failed service.
 		go func() {
@@ -316,12 +332,12 @@ func run(appCallback *AppCallback, stopCh chan struct{}) {
 	mainLog.Load().Info().Msgf("os: %s", osVersion())
 
 	// Wait for network up.
-	if !ctrldnet.Up() {
+	if !networkUp() {
 		failRunUnclassified(mainLog.Load().Error(), "network is not up yet", p.notifyExitToLogServer)
 		return
 	}
 
-	p.router = router.New(&cfg, cdUID != "")
+	p.router = newRouter(&cfg, cdUID != "")
 	cs, err := newControlServer(filepath.Join(sockDir, ControlSocketName()))
 	if err != nil {
 		mainLog.Load().Warn().Err(err).Msg("could not create control server")
@@ -492,6 +508,9 @@ func run(appCallback *AppCallback, stopCh chan struct{}) {
 
 	close(waitCh)
 	<-stopCh
+	if !isMobile() {
+		p.finishRun()
+	}
 }
 
 func writeConfigFile(cfg *ctrld.Config) error {
@@ -2200,8 +2219,9 @@ func newSocketControlClientMobile(dir string, stopCh chan struct{}) *controlClie
 		case <-stopCh:
 			return nil
 		default:
-			_, err := cc.post("/", nil)
+			resp, err := cc.post("/", nil)
 			if err == nil {
+				resp.Body.Close()
 				return cc
 			} else {
 				bo.BackOff(ctx, err)
@@ -2339,6 +2359,7 @@ func checkDeactivationPin(s service.Service, stopCh chan struct{}) error {
 	resp, err := cc.post(deactivationPath, bytes.NewReader(data))
 	mainLog.Load().Debug().Msg("Posting deactivation request done")
 	if resp != nil {
+		defer resp.Body.Close()
 		switch resp.StatusCode {
 		case http.StatusBadRequest:
 			mainLog.Load().Error().Msg(errRequiredDeactivationPin.Error())

cmd/cli/control_server.go

@@ -76,7 +76,6 @@ func (s *controlServer) start() error {
 }
 
 func (s *controlServer) stop() error {
-	_ = os.Remove(s.addr)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
 	defer cancel()
 	return s.server.Shutdown(ctx)

cmd/cli/dns_intercept_darwin.go

@@ -1480,6 +1480,10 @@ func (p *prog) pfStabilizationLoop(ctx context.Context, stableRequired time.Dura
 
 func (p *prog) pfStabilizationLoopWithMaxWait(ctx context.Context, stableRequired, maxWaitDuration time.Duration) {
 	defer p.pfStabilizing.Store(false)
+	if !p.beginNetworkActivity() {
+		return
+	}
+	defer p.netMonitorWG.Done()
 
 	pollInterval := 1500 * time.Millisecond
 	pollTicker := time.NewTicker(pollInterval)
@@ -1504,6 +1508,9 @@ func (p *prog) pfStabilizationLoopWithMaxWait(ctx context.Context, stableRequire
 		case <-pollTicker.C:
 		}
 
+		if p.networkActivityClosed() {
+			return
+		}
 		if p.pfExecBackoffActive() {
 			continue
 		}
@@ -2075,6 +2082,10 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati
 		p.pfSettleFollowupTimer.Stop()
 	}
 	p.pfSettleFollowupTimer = time.AfterFunc(delay, func() {
+		if !p.beginNetworkActivity() {
+			return
+		}
+		defer p.netMonitorWG.Done()
 		if p.dnsInterceptState == nil {
 			return
 		}
@@ -2083,7 +2094,7 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati
 }
 
 // stopPFSettleFollowup cancels a pending post-settle refresh. A callback that already
-// fired still checks dnsInterceptState, so either teardown order is safe.
+// fired is fenced and joined by closeNetMonitor during program shutdown.
 func (p *prog) stopPFSettleFollowup() {
 	p.pfDelayedRecheckMu.Lock()
 	defer p.pfDelayedRecheckMu.Unlock()
@@ -2161,6 +2172,10 @@ func (p *prog) scheduleDelayedRechecks() {
 	for _, delay := range []time.Duration{pfAnchorRecheckDelay, pfAnchorRecheckDelayLong} {
 		delay := delay
 		timer := time.AfterFunc(delay, func() {
+			if !p.beginNetworkActivity() {
+				return
+			}
+			defer p.netMonitorWG.Done()
 			if p.dnsInterceptState == nil || p.pfStabilizing.Load() {
 				return
 			}
@@ -2471,6 +2486,10 @@ var pfInterceptMonitorDelays = []time.Duration{0, 500 * time.Millisecond, time.S
 // The backoff schedule provides both fast detection (immediate + 500ms) and extended
 // coverage (up to ~8s) to win the race against async pf reloads by hypervisors.
 func (p *prog) pfInterceptMonitor() {
+	if !p.beginNetworkActivity() {
+		return
+	}
+	defer p.netMonitorWG.Done()
 	// Eligibility first, ownership second. A monitor that is about to stand down must not
 	// take the flag on its way out: the post-stabilization verifier reads that flag as
 	// "another prober is working" and would step aside for a prober that never probes.
@@ -2494,7 +2513,7 @@ func (p *prog) pfInterceptMonitor() {
 		if delay > 0 {
 			time.Sleep(delay)
 		}
-		if !p.interceptProbeMonitorAllowed() {
+		if p.networkActivityClosed() || !p.interceptProbeMonitorAllowed() {
 			mainLog.Load().Debug().Msg("DNS intercept monitor: aborting — intercept disabled or stabilizing")
 			return
 		}

cmd/cli/dns_intercept_windows.go

@@ -2744,6 +2744,10 @@ func (p *prog) healBlockedLoopbackDNS(state *wfpState, reason string) bool {
 func (p *prog) scheduleDelayedRechecks() {
 	for _, delay := range []time.Duration{pfAnchorRecheckDelay, pfAnchorRecheckDelayLong} {
 		time.AfterFunc(delay, func() {
+			if !p.beginNetworkActivity() {
+				return
+			}
+			defer p.netMonitorWG.Done()
 			if p.dnsInterceptState == nil {
 				return
 			}

cmd/cli/dns_proxy.go

@@ -285,6 +285,7 @@ func (p *prog) serveDNS(listenerNum string) error {
 				defer s.Shutdown()
 				select {
 				case <-p.stopCh:
+				case <-p.runAbortCh:
 				case <-ctx.Done():
 				case err := <-errCh:
 					// Local ipv6 listener should not terminate ctrld.
@@ -305,6 +306,7 @@ func (p *prog) serveDNS(listenerNum string) error {
 						defer s.Shutdown()
 						select {
 						case <-p.stopCh:
+						case <-p.runAbortCh:
 						case <-ctx.Done():
 						case err := <-errCh:
 							// RFC1918 listener should not terminate ctrld.
@@ -321,10 +323,19 @@ func (p *prog) serveDNS(listenerNum string) error {
 			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 <-ctx.Done():
+				return nil
+			}
 
 			select {
 			case <-p.stopCh:
+			case <-p.runAbortCh:
 			case <-ctx.Done():
 			case err := <-errCh:
 				return err
@@ -1279,7 +1290,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 {
@@ -1440,6 +1451,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
 	}
@@ -1456,9 +1471,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
 	})
 }
@@ -1670,20 +1708,35 @@ func FlushDNSCache() error {
 	return nil
 }
 
-// monitorNetworkChanges starts monitoring for network interface changes
+// 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() error {
 	mon, err := newNetworkChangeMonitorFn(func(format string, args ...any) {
-		// Always fetch the latest logger (and inject the prefix)
+		// Always fetch the latest logger (and inject the prefix).
 		mainLog.Load().Printf("netmon: "+format, args...)
 	})
 	if err != nil {
 		return fmt.Errorf("creating network monitor: %w", err)
 	}
-
-	mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) {
+	mon.RegisterChangeCallback(p.networkChangeCallback(func(delta *netmon.ChangeDelta) {
 		p.handleNetworkChange(delta, mon.IsMajorChangeFrom(delta.Old, delta.New))
-	})
-
+	}))
+	if !p.setNetMonitor(mon) {
+		_ = mon.Close()
+		mainLog.Load().Debug().Msg("network monitor discarded, ctrld is stopping")
+		return nil
+	}
 	mon.Start()
 	mainLog.Load().Debug().Msg("Network monitor started")
 	return nil
@@ -2222,12 +2275,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()
 		mainLog.Load().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()
@@ -2293,6 +2353,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()
@@ -2313,6 +2377,15 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u
 	targetBefore := p.interceptTargetSnapshot()
 	outcome := recoveryOutcomeSuperseded
 	defer func() { p.endRecovery(&diagnostic, outcome, targetBefore) }()
+	// Clean every exit, including shutdown before the first DNS mutation.
+	// Run cleanup before the end snapshot so it reports the final state.
+	defer p.recoveryCanceledCleanup(gen)
+	if recoveryCtx.Err() != nil {
+		if p.recoveryGen.Load() == gen {
+			outcome = recoveryOutcomeCanceled
+		}
+		return
+	}
 	if reason == RecoveryReasonNetworkChange {
 		mainLog.Load().Debug().Msg("Network change recovery now owns shared recovery state")
 	}
@@ -2399,10 +2472,9 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u
 	// Wait indefinitely until one of the upstreams recovers.
 	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
@@ -2482,6 +2554,10 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 	var wg sync.WaitGroup
 
 	mainLog.Load().Debug().Msgf("Starting upstream recovery check for %d upstreams", len(upstreams))
+	defer func() {
+		cancel()
+		wg.Wait()
+	}()
 
 	for name, uc := range upstreams {
 		wg.Add(1)
@@ -2561,7 +2637,6 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 	case <-ctx.Done():
 		return "", ctx.Err()
 	}
-	wg.Wait()
 	return recovered, nil
 }
 

cmd/cli/force_reload_lifetime_test.go

@@ -0,0 +1,48 @@
+package cli
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestForceFetchingAPIDrainsBeforeRestoration(t *testing.T) {
+	old := cdUID
+	cdUID = "testuid"
+	t.Cleanup(func() { cdUID = old })
+	p := &prog{cfg: &ctrld.Config{}, stopCh: make(chan struct{}), runAbortCh: make(chan struct{}), apiForceReloadCh: make(chan struct{})}
+	// Pause the admitted producer after delivery, before it reads cooldown
+	// configuration. Shutdown must join it even though the consumer is done.
+	p.mu.Lock()
+	locked := true
+	defer func() {
+		if locked {
+			p.mu.Unlock()
+		}
+	}()
+	p.forceFetchingAPI("testuid.verify.controld.com")
+	select {
+	case <-p.apiForceReloadCh:
+	case <-time.After(time.Second):
+		t.Fatal("force-reload producer did not send")
+	}
+	closed := make(chan struct{})
+	go func() { p.closeNetMonitor(); close(closed) }()
+	waitNetworkShutdownGate(t, p)
+	assertNetworkShutdownBlocked(t, closed)
+	p.mu.Unlock()
+	locked = false
+	waitNetworkShutdownSignal(t, closed, "force-reload producer drain")
+	// Service restoration precedes closing stopCh, so that channel alone
+	// cannot cancel the producer's send or cooldown while restoration joins it.
+	if stopRequested(p.stopCh) || stopRequested(p.runAbortCh) {
+		t.Fatal("network shutdown closed a caller-owned lifetime channel")
+	}
+	p.forceFetchingAPI("testuid.verify.controld.com")
+	select {
+	case <-p.apiForceReloadCh:
+		t.Fatal("shutdown admitted a late force-reload producer")
+	default:
+	}
+}

cmd/cli/internal_log_shutdown_test.go

@@ -0,0 +1,50 @@
+package cli
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestFinishRunClosesInternalFilesWithoutServiceStop(t *testing.T) {
+	p := &prog{
+		cfg: &ctrld.Config{}, runDone: make(chan struct{}), runAbortCh: make(chan struct{}),
+		dnsWatcherStopCh:  make(chan struct{}),
+		internalLogWriter: newLogWriter(), internalJournalWriter: newSmallLogWriter(),
+	}
+	var files []*os.File
+	var paths []string
+	for i, writer := range []*logWriter{p.internalLogWriter, p.internalJournalWriter} {
+		path := filepath.Join(t.TempDir(), []string{"debug.log", "journal.log"}[i])
+		if err := writer.setLogFile(path, logBudget{maxSize: 4096, backups: 1}); err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(writer.closeLogFile)
+		files = append(files, writer.rotating().f)
+		paths = append(paths, path)
+	}
+	p.onStopped = []func(){func() {
+		for _, writer := range []*logWriter{p.internalLogWriter, p.internalJournalWriter} {
+			if _, err := writer.Write([]byte("final cleanup\n")); err != nil {
+				t.Error(err)
+			}
+		}
+	}}
+	close(p.runDone)
+	p.finishRun()
+	for i, f := range files {
+		if _, err := f.Stat(); !errors.Is(err, os.ErrClosed) {
+			t.Errorf("file %d is not closed: %v", i, err)
+		}
+		b, err := os.ReadFile(paths[i])
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(b) != "final cleanup\n" {
+			t.Errorf("file %d lost final cleanup: %q", i, b)
+		}
+	}
+}

cmd/cli/listener_shutdown_test.go

@@ -0,0 +1,293 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net"
+	"os"
+	"os/exec"
+	"runtime"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/miekg/dns"
+	"github.com/stretchr/testify/require"
+
+	"github.com/Control-D-Inc/ctrld"
+	"github.com/Control-D-Inc/ctrld/internal/controld"
+)
+
+func TestStopRestorationFailureStillStopsWorkers(t *testing.T) {
+	old := deAllocateIPFn
+	t.Cleanup(func() { deAllocateIPFn = old })
+	want := errors.New("deallocation failed")
+	calls := 0
+	deAllocateIPFn = func(string) error { calls++; return want }
+	p := &prog{
+		cfg:    &ctrld.Config{Service: ctrld.ServiceConfig{AllocateIP: true}, Listener: map[string]*ctrld.ListenerConfig{"0": {IP: "127.0.0.1"}}},
+		waitCh: make(chan struct{}), stopCh: make(chan struct{}), runDone: make(chan struct{}),
+		runAbortCh: make(chan struct{}), dnsWatcherStopCh: make(chan struct{}),
+		pinCodeValidCh: make(chan struct{}, 1),
+	}
+	// Accept the stop even if another test configured a deactivation pin.
+	p.pinCodeValidCh <- struct{}{}
+	require.NoError(t, p.Start(nil))
+	require.ErrorIs(t, p.Stop(nil), want)
+	require.True(t, stopRequested(p.stopCh))
+	awaitShutdown(t, p.runDone)
+	p.finishRun()
+	require.ErrorIs(t, p.restoreOSState(), want, "the first error must survive sync.Once")
+	require.Equal(t, 1, calls, "never invoke the real IP deallocator")
+}
+
+func awaitShutdown(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("shutdown did not complete")
+	}
+}
+
+func TestFinishRunWaitsBeyondTimeout(t *testing.T) {
+	c, peer := net.Pipe()
+	defer peer.Close()
+	p := &prog{cfg: &ctrld.Config{}, runAbortCh: make(chan struct{}), runDone: make(chan struct{}), dnsWatcherStopCh: make(chan struct{}), logConn: c}
+	done := make(chan struct{})
+	go func() { p.finishRun(); close(done) }()
+	select {
+	case <-done:
+		t.Fatal("released resources while the run was alive")
+	case <-time.After(shutdownTimeout + 50*time.Millisecond):
+	}
+	close(p.runDone)
+	awaitShutdown(t, done)
+	_, err := peer.Write([]byte("x"))
+	require.ErrorIs(t, err, io.ErrClosedPipe)
+}
+
+// Each case runs in a subprocess because run initializes package-global config
+// and logging. No service, interface DNS or network-monitor setup is permitted.
+func TestRunListenerShutdown(t *testing.T) {
+	const env = "CTRLD_TEST_LISTENER_SHUTDOWN"
+	mode := os.Getenv(env)
+	if mode == "" {
+		for _, mode := range []string{"stop-before-ready", "abort-before-ready", "drain", "reload", "reload-before-ready", "stop-during-hook", "stop-during-post", "wire-udp", "wire-tcp", "api-drain"} {
+			t.Run(mode, func(t *testing.T) {
+				ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second)
+				defer cancel()
+				cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestRunListenerShutdown$", "-test.v")
+				cmd.Env = append(os.Environ(), env+"="+mode)
+				out, err := cmd.CombinedOutput()
+				require.NoError(t, err, "%s", out)
+			})
+		}
+		return
+	}
+	defer func() {
+		if t.Failed() {
+			t.Log(logOutput.String())
+			buf := make([]byte, 1<<20)
+			n := runtime.Stack(buf, true)
+			t.Log(string(buf[:n]))
+		}
+	}()
+	isMobile = func() bool { return true }
+	cdUID = ""
+	iface = ""
+	apiEntered, apiCanceled, apiRelease := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	if mode == "api-drain" {
+		cdUID = "testuid"
+		fetchResolverConfigForReload = func(ctx context.Context, _ *controld.ResolverConfigRequest, _ bool) (*controld.ResolverConfig, error) {
+			close(apiEntered)
+			<-ctx.Done()
+			close(apiCanceled)
+			<-apiRelease
+			return nil, ctx.Err()
+		}
+	}
+	var monitorCalls, postCalls atomic.Int32
+	monitorNetworkChangesFn = func(*prog) error { monitorCalls.Add(1); return nil }
+	postRunFn = func(*prog) { postCalls.Add(1) }
+	dir := t.TempDir()
+	cfg = ctrld.Config{Service: ctrld.ServiceConfig{LogPath: dir + "/ctrld.log"}, Listener: map[string]*ctrld.ListenerConfig{"0": {IP: "127.0.0.1"}}}
+	p := &prog{
+		cfg: &cfg, waitCh: make(chan struct{}), stopCh: make(chan struct{}), runAbortCh: make(chan struct{}),
+		runDone: make(chan struct{}), dnsWatcherStopCh: make(chan struct{}), apiReloadCh: make(chan *ctrld.Config),
+		reloadDoneCh:     make(chan struct{}, 1),
+		apiForceReloadCh: make(chan struct{}),
+	}
+	logConnection := &closeCountingLogConn{}
+	p.logConn = logConnection
+	entered, drain, release := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	defer unblock()
+	var hooks atomic.Int32
+	p.onStarted = []func(){func() { hooks.Add(1) }}
+	mutationEntered, mutationRelease, restored := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	stopDuringMutation := mode == "stop-during-hook" || mode == "stop-during-post"
+	if stopDuringMutation {
+		mutation := func() {
+			close(mutationEntered)
+			<-mutationRelease
+			select {
+			case <-restored:
+				t.Error("startup mutation ran after restoration")
+			default:
+			}
+		}
+		p.onStopped = []func(){func() { close(restored) }}
+		if mode == "stop-during-hook" {
+			p.onStarted = []func(){mutation}
+		} else {
+			postRunFn = func(*prog) { mutation(); postCalls.Add(1) }
+		}
+	}
+	wire := mode == "wire-udp" || mode == "wire-tcp"
+	if wire {
+		ln, err := net.Listen("tcp", "127.0.0.1:0")
+		require.NoError(t, err)
+		cfg.Listener["0"].Port = ln.Addr().(*net.TCPAddr).Port
+		require.NoError(t, ln.Close())
+		// The real handler blocks in its semaphore while both server Shutdown
+		// calls drain. Use an internal query so no external upstream is needed.
+		serveDNSFn = func(p *prog, n string) error {
+			p.sema = &shutdownSemaphore{entered: entered, resume: release}
+			return p.serveDNS(n)
+		}
+	} else {
+		serveDNSFn = func(p *prog, _ string) error {
+			close(entered)
+			if mode == "drain" || mode == "reload" || mode == "api-drain" || stopDuringMutation {
+				p.started <- struct{}{}
+			}
+			select {
+			case <-p.stopCh:
+			case <-p.runAbortCh:
+			}
+			close(drain)
+			<-release
+			return nil
+		}
+	}
+	close(p.waitCh)
+	require.NoError(t, p.Start(nil))
+	var queryDone chan error
+	if wire {
+		require.Eventually(t, func() bool { return postCalls.Load() == 1 }, time.Second, time.Millisecond)
+		proto := mode[len("wire-"):]
+		queryDone = make(chan error, 1)
+		go func() {
+			m := new(dns.Msg)
+			m.SetQuestion(selfCheckInternalTestDomain+".", dns.TypeA)
+			answer, _, err := (&dns.Client{Net: proto, Timeout: 5 * time.Second}).Exchange(m, net.JoinHostPort("127.0.0.1", strconv.Itoa(cfg.Listener["0"].Port)))
+			if err == nil && answer.Rcode != dns.RcodeSuccess {
+				err = errors.New("internal query failed")
+			}
+			queryDone <- err
+		}()
+	}
+	awaitShutdown(t, entered)
+	if mode == "api-drain" {
+		select {
+		case p.apiForceReloadCh <- struct{}{}:
+		case <-time.After(time.Second):
+			t.Fatal("production run did not start the API reload worker")
+		}
+		awaitShutdown(t, apiEntered)
+	}
+	if mode == "reload" || mode == "reload-before-ready" {
+		if mode == "reload" {
+			require.Eventually(t, func() bool { return postCalls.Load() == 1 }, time.Second, time.Millisecond)
+		}
+		configPath = dir + "/reload.toml"
+		p.apiReloadCh <- &ctrld.Config{
+			Service:  cfg.Service,
+			Listener: map[string]*ctrld.ListenerConfig{"0": {IP: "127.0.0.1", Port: 5354}},
+			Network:  map[string]*ctrld.NetworkConfig{"0": {Cidrs: []string{"127.0.0.0/8"}}},
+			Upstream: map[string]*ctrld.UpstreamConfig{"0": {Type: ctrld.ResolverTypeDOT, Endpoint: "example.test:853", BootstrapIP: "127.0.0.1"}},
+		}
+		if mode == "reload-before-ready" {
+			// Give the reload a chance to reach the old run's cancellation path.
+			select {
+			case <-p.reloadDoneCh:
+				t.Fatal("reload skipped one-time startup")
+			case <-time.After(30 * time.Millisecond):
+			}
+			p.started <- struct{}{}
+		}
+		select {
+		case <-p.reloadDoneCh:
+		case <-time.After(3 * time.Second):
+			t.Fatal("reload waited for a persistent listener to stop")
+		}
+		select {
+		case <-drain:
+			t.Fatal("reload stopped the persistent listener")
+		default:
+		}
+		require.EqualValues(t, 1, hooks.Load())
+		require.EqualValues(t, 1, postCalls.Load())
+	}
+	if stopDuringMutation {
+		awaitShutdown(t, mutationEntered)
+		stopDone := make(chan struct{})
+		go func() { _ = p.Stop(nil); close(stopDone) }()
+		select {
+		case <-restored:
+			t.Fatal("restoration raced a startup OS mutation")
+		case <-time.After(30 * time.Millisecond):
+		}
+		close(mutationRelease)
+		awaitShutdown(t, stopDone)
+		awaitShutdown(t, restored)
+		require.False(t, p.startOSState(func() { t.Error("startup permitted after restoration") }))
+	} else if mode != "abort-before-ready" {
+		close(p.stopCh)
+	}
+	finished := make(chan struct{})
+	go func() { p.finishRun(); close(finished) }()
+	awaitShutdown(t, p.runDone)
+	if !wire {
+		awaitShutdown(t, drain)
+	}
+	select {
+	case <-finished:
+		t.Fatal("finishRun returned before the listener/handler drained")
+	case <-time.After(30 * time.Millisecond):
+	}
+	unblock()
+	if mode == "api-drain" {
+		awaitShutdown(t, apiCanceled)
+		select {
+		case <-finished:
+			t.Fatal("finishRun returned before the API reload worker")
+		case <-time.After(30 * time.Millisecond):
+		}
+		close(apiRelease)
+	}
+	awaitShutdown(t, finished)
+	require.EqualValues(t, 1, logConnection.closes.Load(), "startup and shutdown must share log-connection ownership")
+	if wire {
+		require.NoError(t, <-queryDone, "active DNS query must finish before resource retirement")
+	}
+	if mode == "stop-before-ready" || mode == "abort-before-ready" {
+		require.Zero(t, hooks.Load(), "startup hooks must not run after cancellation")
+		require.Zero(t, postCalls.Load())
+	}
+	require.EqualValues(t, 1, monitorCalls.Load(), "real network monitor must not be reached")
+}
+
+type shutdownSemaphore struct {
+	entered chan struct{}
+	resume  <-chan struct{}
+	once    sync.Once
+}
+
+func (s *shutdownSemaphore) acquire() { s.once.Do(func() { close(s.entered) }); <-s.resume }
+func (*shutdownSemaphore) release()   {}

cmd/cli/network_monitor_test.go

@@ -13,6 +13,8 @@ type sourceTestMonitor struct {
 	started  bool
 }
 
+func (*sourceTestMonitor) Close() error { return nil }
+
 func (m *sourceTestMonitor) RegisterChangeCallback(fn netmon.ChangeFunc) func() {
 	m.callback = fn
 	return func() {}

cmd/cli/network_shutdown_test.go

@@ -0,0 +1,322 @@
+package cli
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+func guardNetworkShutdownHostMutations(t *testing.T) {
+	t.Helper()
+	oldReset, oldRemove := recoveryResetDNSFn, removeInterceptDNSTargetFn
+	recoveryResetDNSFn = func(*prog, bool, bool) { panic("unexpected recovery DNS reset") }
+	// Also fence the first platform boundary inside real resetDNS, so mutation
+	// testing a bypassed recoveryResetDNSFn can never touch the test runner's DNS/PF.
+	removeInterceptDNSTargetFn = func(*prog, string) { panic("real resetDNS crossed host safety boundary") }
+	t.Cleanup(func() { recoveryResetDNSFn, removeInterceptDNSTargetFn = oldReset, oldRemove })
+}
+
+func waitNetworkShutdownSignal(t *testing.T, ch <-chan struct{}, what string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for %s", what)
+	}
+}
+
+func waitNetworkShutdownGate(t *testing.T, p *prog) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for !p.networkActivityClosed() {
+		if time.Now().After(deadline) {
+			t.Fatal("shutdown did not close admission")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func assertNetworkShutdownBlocked(t *testing.T, ch <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-ch:
+		t.Fatal("shutdown returned before admitted work finished")
+	case <-time.After(20 * time.Millisecond):
+	}
+}
+
+func TestNetworkShutdownRejectsLateMonitorPublication(t *testing.T) {
+	p := &prog{}
+	p.closeNetMonitor()
+	// A zero monitor is never started or closed: on rejected publication the
+	// initializer retains ownership. No OS monitor or DNS policy is installed.
+	if p.setNetMonitor(new(netmon.Monitor)) {
+		t.Fatal("monitor published after shutdown")
+	}
+	if p.netMonitor != nil {
+		t.Fatal("shutdown retained a monitor")
+	}
+	if p.beginNetworkActivity() {
+		p.netMonitorWG.Done()
+		t.Fatal("work admitted after shutdown")
+	}
+}
+
+func TestNetworkShutdownDrainsCallbacksAndConcurrentClosers(t *testing.T) {
+	guardNetworkShutdownHostMutations(t)
+	p := &prog{}
+	entered, release := make(chan struct{}), make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	defer unblock()
+	var calls atomic.Int32
+	// Stub the entire callback body before any platform or DNS boundary. This
+	// is the same wrapper used by RegisterChangeCallback in production.
+	callback := p.networkChangeCallback(func(*netmon.ChangeDelta) {
+		calls.Add(1)
+		close(entered)
+		<-release
+		// An admitted callback may try to schedule recovery after Close fenced
+		// admission. Neither synchronous nor deferred work may start then.
+		p.debounceRecovery(p.networkAcceptedGen.Load())
+		p.handleRecovery(RecoveryReasonNetworkChange)
+	})
+	go callback(nil)
+	waitNetworkShutdownSignal(t, entered, "callback admission")
+
+	closed1, closed2 := make(chan struct{}), make(chan struct{})
+	go func() { p.closeNetMonitor(); close(closed1) }()
+	waitNetworkShutdownGate(t, p)
+	go func() { p.closeNetMonitor(); close(closed2) }()
+	// Model a callback dispatched by netmon before Close but scheduled later.
+	callback(nil)
+	assertNetworkShutdownBlocked(t, closed1)
+	assertNetworkShutdownBlocked(t, closed2)
+	unblock()
+	waitNetworkShutdownSignal(t, closed1, "first close")
+	waitNetworkShutdownSignal(t, closed2, "second close")
+	callback(nil)
+	if calls.Load() != 1 || p.recoveryGen.Load() != 0 || p.recoveryDebounceTimer != nil {
+		t.Fatalf("late work ran: callbacks=%d recovery=%d timer=%v", calls.Load(), p.recoveryGen.Load(), p.recoveryDebounceTimer)
+	}
+}
+
+func TestNetworkShutdownCancelsAndDrainsRecovery(t *testing.T) {
+	testNetworkShutdownCancelsAndDrainsRecovery(t, func(p *prog) {
+		p.handleRecovery(RecoveryReasonRegularFailure)
+	})
+}
+
+func TestNetworkShutdownDrainsDeferredRecovery(t *testing.T) {
+	testNetworkShutdownCancelsAndDrainsRecovery(t, func(p *prog) {
+		p.networkChangeCallback(func(*netmon.ChangeDelta) { p.debounceRecovery(p.networkAcceptedGen.Load()) })(nil)
+	})
+}
+
+func testNetworkShutdownCancelsAndDrainsRecovery(t *testing.T, start func(*prog)) {
+	logs := captureDebugMainLog(t)
+	stubHeaderSnapshotSources(t)
+	guardNetworkShutdownHostMutations(t)
+	oldReset, oldIntercept := recoveryResetDNSFn, dnsIntercept
+	dnsIntercept = false
+	defer func() { recoveryResetDNSFn, dnsIntercept = oldReset, oldIntercept }()
+	p := &prog{cfg: &ctrld.Config{}}
+	entered, canceled, release := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	defer unblock()
+	var resets atomic.Int32
+	// resetDNS is the FIRST host-mutating boundary in traditional recovery.
+	// Block there; never invoke real resetDNS, setDNS, PF, or NRPT operations.
+	recoveryResetDNSFn = func(p *prog, isStart, restoreStatic bool) {
+		resets.Add(1)
+		p.recoveryCancelMu.Lock()
+		cancel := p.recoveryCancel
+		p.recoveryCancel = func() { cancel(); close(canceled) }
+		p.recoveryCancelMu.Unlock()
+		close(entered)
+		<-release
+	}
+	recovered := make(chan struct{})
+	go func() { start(p); close(recovered) }()
+	waitNetworkShutdownSignal(t, entered, "recovery reset seam")
+	closed := make(chan struct{})
+	go func() { p.closeNetMonitor(); close(closed) }()
+	waitNetworkShutdownSignal(t, canceled, "recovery cancellation")
+	assertNetworkShutdownBlocked(t, closed)
+	if p.recoveryOwnsState(p.recoveryGen.Load()) {
+		t.Fatal("shutdown recovery still owns permission to reapply DNS")
+	}
+	unblock()
+	waitNetworkShutdownSignal(t, closed, "recovery drain")
+	waitNetworkShutdownSignal(t, recovered, "recovery return")
+	p.handleRecovery(RecoveryReasonRegularFailure)
+	p.handleRecovery(RecoveryReasonNetworkChange)
+	if resets.Load() != 1 || p.recoveryRunning.Load() || p.recoveryBypass.Load() || p.recoveryCancel != nil {
+		t.Fatalf("recovery survived close: resets=%d running=%v bypass=%v", resets.Load(), p.recoveryRunning.Load(), p.recoveryBypass.Load())
+	}
+	ends := jsonLogEvents(t, logs, "Recovery end")
+	if len(ends) != 1 {
+		t.Fatalf("recovery end events: got %d, want 1", len(ends))
+	}
+	wantField(t, ends[0], "outcome", recoveryOutcomeCanceled)
+	wantField(t, ends[0], "bypass_active", false)
+}
+
+func TestNetworkShutdownCancelsIndefiniteRecoveryWait(t *testing.T) {
+	guardNetworkShutdownHostMutations(t)
+	oldReset, oldIntercept := recoveryResetDNSFn, dnsIntercept
+	dnsIntercept = false
+	defer func() { recoveryResetDNSFn, dnsIntercept = oldReset, oldIntercept }()
+	p := &prog{cfg: &ctrld.Config{}}
+	entered := make(chan struct{})
+	recoveryResetDNSFn = func(*prog, bool, bool) { close(entered) }
+	recovered := make(chan struct{})
+	// An empty upstream map can never recover. Shutdown must cancel that wait
+	// without waiting on a mutex held for the recovery's entire lifetime.
+	go func() { p.handleRecovery(RecoveryReasonRegularFailure); close(recovered) }()
+	waitNetworkShutdownSignal(t, entered, "recovery reset seam")
+	closed := make(chan struct{})
+	go func() { p.closeNetMonitor(); close(closed) }()
+	waitNetworkShutdownSignal(t, closed, "canceled indefinite recovery")
+	waitNetworkShutdownSignal(t, recovered, "recovery return")
+}
+
+func TestNetworkShutdownStopsDeferredRecovery(t *testing.T) {
+	guardNetworkShutdownHostMutations(t)
+	p := &prog{}
+	p.debounceRecovery(p.networkAcceptedGen.Load())
+	p.recoveryDebounceMu.Lock()
+	timer := p.recoveryDebounceTimer
+	p.recoveryDebounceMu.Unlock()
+	if timer == nil {
+		t.Fatal("debounce did not schedule recovery")
+	}
+	p.closeNetMonitor()
+	if timer.Stop() {
+		t.Fatal("shutdown did not stop pending debounce timer")
+	}
+	// Model the already-fired timer's call to handleRecovery, which Stop
+	// cannot retract, then a callback trying to create another timer.
+	p.handleRecovery(RecoveryReasonNetworkChange)
+	p.debounceRecovery(p.networkAcceptedGen.Load())
+	if p.recoveryDebounceTimer != nil || p.recoveryGen.Load() != 0 {
+		t.Fatal("deferred recovery restarted after shutdown")
+	}
+}
+
+func TestNetworkShutdownRejectsRecoveryAdmittedBeforeClose(t *testing.T) {
+	p := &prog{}
+	// Pause between handleRecovery's activity admission and beginRecovery.
+	if !p.beginNetworkActivity() {
+		t.Fatal("initial admission failed")
+	}
+	closed := make(chan struct{})
+	go func() { p.closeNetMonitor(); close(closed) }()
+	waitNetworkShutdownGate(t, p)
+	_, _, _, ok := p.beginRecovery(RecoveryReasonNetworkChange)
+	p.netMonitorWG.Done()
+	waitNetworkShutdownSignal(t, closed, "admitted recovery drain")
+	if ok || p.recoveryCancel != nil || p.recoveryRunning.Load() {
+		t.Fatal("recovery published a new owner after shutdown cancellation")
+	}
+}
+
+type shutdownTestMonitor struct {
+	callback       netmon.ChangeFunc
+	starts, closes int
+}
+
+func (m *shutdownTestMonitor) Start()       { m.starts++ }
+func (m *shutdownTestMonitor) Close() error { m.closes++; return nil }
+func (m *shutdownTestMonitor) RegisterChangeCallback(fn netmon.ChangeFunc) func() {
+	m.callback = fn
+	return func() {}
+}
+func (*shutdownTestMonitor) IsMajorChangeFrom(*netmon.State, *netmon.State) bool {
+	panic("closed callback reached network discovery")
+}
+
+func TestNetworkMonitorProductionPublicationAndCallbackFence(t *testing.T) {
+	old := newNetworkChangeMonitorFn
+	t.Cleanup(func() { newNetworkChangeMonitorFn = old })
+	for _, late := range []bool{false, true} {
+		p := &prog{}
+		m := &shutdownTestMonitor{}
+		newNetworkChangeMonitorFn = func(func(string, ...any)) (networkChangeMonitor, error) {
+			if late {
+				p.closeNetMonitor()
+			}
+			return m, nil
+		}
+		if err := p.monitorNetworkChanges(); err != nil {
+			t.Fatal(err)
+		}
+		p.closeNetMonitor()
+		if m.closes != 1 {
+			t.Fatalf("monitor closed %d times", m.closes)
+		}
+		wantStarts := 1
+		if late {
+			wantStarts = 0
+		}
+		if m.starts != wantStarts {
+			t.Fatalf("monitor starts=%d want=%d", m.starts, wantStarts)
+		}
+		// Exercise the callback captured from the actual production registration.
+		// Removing its wrapper reaches the panic above, before any OS mutation.
+		m.callback(&netmon.ChangeDelta{})
+	}
+}
+
+func TestFinishRunDrainsNetworkJournalOnAbort(t *testing.T) {
+	captureDebugMainLog(t)
+	stubHeaderSnapshotSources(t)
+	p := &prog{
+		cfg: &ctrld.Config{}, stopCh: make(chan struct{}),
+		runAbortCh: make(chan struct{}), runDone: make(chan struct{}),
+		dnsWatcherStopCh: make(chan struct{}),
+	}
+	p.newNetworkJournal()
+	entered, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	unblock := func() { once.Do(func() { close(release) }) }
+	defer unblock()
+	// The poll is a synthetic first platform boundary, even on Darwin.
+	p.dnsConfig = newDNSConfigPoller(func() ([]dnsResolverEntry, error) {
+		close(entered)
+		<-release
+		return nil, nil
+	})
+	p.waitNetworkJournal = p.startNetworkJournal()
+	waitNetworkShutdownSignal(t, entered, "journal poll")
+	close(p.runDone)
+	done := make(chan struct{})
+	go func() { p.finishRun(); close(done) }()
+	waitNetworkShutdownSignal(t, p.runAbortCh, "run abort")
+	assertNetworkShutdownBlocked(t, done)
+	unblock()
+	waitNetworkShutdownSignal(t, done, "journal drain on abort")
+	if stopRequested(p.stopCh) {
+		t.Fatal("finishRun closed the caller-owned stop channel")
+	}
+}
+
+func TestNetworkShutdownDrainsBeforeOSRestoration(t *testing.T) {
+	p := &prog{cfg: &ctrld.Config{}, dnsWatcherStopCh: make(chan struct{})}
+	entered, release, restored := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	p.onStopped = []func(){func() { close(restored) }}
+	go p.networkChangeCallback(func(*netmon.ChangeDelta) { close(entered); <-release })(nil)
+	waitNetworkShutdownSignal(t, entered, "callback entry")
+	done := make(chan struct{})
+	go func() { _ = p.restoreOSState(); close(done) }()
+	waitNetworkShutdownGate(t, p)
+	assertNetworkShutdownBlocked(t, restored)
+	close(release)
+	waitNetworkShutdownSignal(t, done, "restoration")
+	waitNetworkShutdownSignal(t, restored, "restore hook")
+}

cmd/cli/network_transition.go

@@ -17,6 +17,7 @@ import (
 )
 
 type networkChangeMonitor interface {
+	Close() error
 	RegisterChangeCallback(netmon.ChangeFunc) func()
 	IsMajorChangeFrom(*netmon.State, *netmon.State) bool
 	Start()

cmd/cli/prog.go

@@ -113,6 +113,7 @@ type prog struct {
 	apiForceReloadCh     chan struct{}
 	apiForceReloadGroup  singleflight.Group
 	logConn              net.Conn
+	logConnMu            sync.Mutex
 	cs                   *controlServer
 	csSetDnsDone         chan struct{}
 	csSetDnsOk           bool
@@ -321,6 +322,111 @@ type prog struct {
 	onStartedDone chan struct{}
 	onStarted     []func()
 	onStopped     []func()
+
+	netMonitorMu     sync.Mutex
+	netMonitor       networkChangeMonitor
+	netMonitorClosed bool
+	netMonitorStopCh chan struct{}
+	restoreOnce      sync.Once
+	restoreErr       error
+	releaseOnce      sync.Once
+	runDone          chan struct{}
+	runAbortCh       chan struct{}
+	listenerWg       sync.WaitGroup
+	apiReloadWg      sync.WaitGroup
+	osStateMu        sync.Mutex
+	osStateRestored  bool
+
+	// netMonitorWG tracks admitted callbacks and their recovery work. Admission
+	// is serialized with shutdown under netMonitorMu; no Add races with Wait.
+	netMonitorWG        sync.WaitGroup
+	netMonitorCloseOnce sync.Once
+	// Published by the joined initializer; read only after runDone.
+	waitNetworkJournal func()
+}
+
+// setNetMonitor publishes mon as the active network monitor, closing any
+// predecessor. It reports false if shutdown already ran, in which case the
+// caller owns mon and must not start it.
+func (p *prog) setNetMonitor(mon networkChangeMonitor) bool {
+	p.netMonitorMu.Lock()
+	defer p.netMonitorMu.Unlock()
+	if p.netMonitorClosed {
+		return false
+	}
+	old := p.netMonitor
+	p.netMonitor = mon
+	if old != nil {
+		_ = old.Close()
+	}
+	return true
+}
+
+// beginNetworkActivity admits a callback or recovery before shutdown. Every
+// successful admission must be paired with netMonitorWG.Done.
+func (p *prog) beginNetworkActivity() bool {
+	p.netMonitorMu.Lock()
+	defer p.netMonitorMu.Unlock()
+	if p.netMonitorClosed {
+		return false
+	}
+	p.netMonitorWG.Add(1)
+	return true
+}
+
+func (p *prog) networkActivityClosed() bool {
+	p.netMonitorMu.Lock()
+	defer p.netMonitorMu.Unlock()
+	return p.netMonitorClosed
+}
+
+// networkActivityDone cancels admitted work before service Stop closes stopCh.
+func (p *prog) networkActivityDone() <-chan struct{} {
+	p.netMonitorMu.Lock()
+	defer p.netMonitorMu.Unlock()
+	if p.netMonitorStopCh == nil {
+		p.netMonitorStopCh = make(chan struct{})
+		if p.netMonitorClosed {
+			close(p.netMonitorStopCh)
+		}
+	}
+	return p.netMonitorStopCh
+}
+
+// closeNetMonitor fences publications and callbacks, cancels recovery, and
+// drains admitted work before OS state restoration. netmon.Close alone does
+// not join change callbacks. Never hold an admission/state mutex while waiting:
+// callbacks and canceled recoveries need those mutexes to finish.
+func (p *prog) closeNetMonitor() {
+	p.netMonitorCloseOnce.Do(func() {
+		p.netMonitorMu.Lock()
+		mon := p.netMonitor
+		p.netMonitor = nil
+		p.netMonitorClosed = true
+		if p.netMonitorStopCh != nil {
+			close(p.netMonitorStopCh)
+		}
+		p.netMonitorMu.Unlock()
+
+		p.recoveryDebounceMu.Lock()
+		if p.recoveryDebounceTimer != nil {
+			p.recoveryDebounceTimer.Stop()
+			p.recoveryDebounceTimer = nil
+		}
+		p.recoveryDebounceMu.Unlock()
+
+		p.recoveryCancelMu.Lock()
+		if p.recoveryCancel != nil {
+			p.recoveryCancel()
+		}
+		p.recoveryCancelMu.Unlock()
+
+		if mon != nil {
+			_ = mon.Close()
+			mainLog.Load().Debug().Msg("network monitor stopped")
+		}
+		p.netMonitorWG.Wait()
+	})
 }
 
 func (p *prog) Start(s service.Service) error {
@@ -330,10 +436,11 @@ func (p *prog) Start(s service.Service) error {
 
 // runWait runs ctrld components, waiting for signal to reload.
 func (p *prog) runWait() {
-	p.mu.Lock()
-	p.cfg = &cfg
-	p.mu.Unlock()
+	if p.runDone != nil {
+		defer close(p.runDone)
+	}
 	reloadSigCh := make(chan os.Signal, 1)
+	defer stopNotifyReloadSigCh(reloadSigCh)
 	notifyReloadSigCh(reloadSigCh)
 
 	reload := false
@@ -357,6 +464,11 @@ func (p *prog) runWait() {
 			newCfg = apiCfg
 		case <-p.stopCh:
 			close(reloadCh)
+			<-done
+			return
+		case <-p.runAbortCh:
+			close(reloadCh)
+			<-done
 			return
 		}
 
@@ -436,6 +548,7 @@ func (p *prog) runWait() {
 		p.setupUpstream(newCfg)
 
 		p.mu.Lock()
+		oldUpstreams := p.cfg.Upstream
 		*p.cfg = *newCfg
 		// In DNS-intercept mode on macOS, the DNS listener is bound once at startup and is
 		// NOT re-bound on reload (see prog.run: serveDNS is started only when !reload). When
@@ -451,6 +564,7 @@ func (p *prog) runWait() {
 		}
 		p.mu.Unlock()
 
+		closeReplacedUpstreams(oldUpstreams, newCfg.Upstream)
 		// The header names the mode, the listeners, and the upstreams, so the
 		// open files take the values of the config that now runs.
 		p.refreshLogHeader()
@@ -526,13 +640,42 @@ func (p *prog) postRun() {
 	}
 }
 
+var fetchResolverConfigForReload = controld.FetchResolverConfig
+
+// startAPIConfigReload starts one program-lifetime worker, not one per reload.
+func (p *prog) startAPIConfigReload() {
+	p.apiReloadWg.Add(1)
+	go func() {
+		defer p.apiReloadWg.Done()
+		p.apiConfigReload()
+	}()
+}
+
 // apiConfigReload calls API to check for latest config update then reload ctrld if necessary.
 func (p *prog) apiConfigReload() {
 	if cdUID == "" {
 		return
 	}
+	ctx, cancel := context.WithCancel(context.Background())
+	watcherDone := make(chan struct{})
+	defer func() {
+		cancel()
+		<-watcherDone
+	}()
+	go func() {
+		defer close(watcherDone)
+		select {
+		case <-p.stopCh:
+		case <-p.runAbortCh:
+		case <-ctx.Done():
+		}
+		cancel()
+	}()
 
-	ticker := time.NewTicker(timeDurationOrDefault(p.cfg.Service.RefetchTime, 3600) * time.Second)
+	p.mu.Lock()
+	refetchInterval := timeDurationOrDefault(p.cfg.Service.RefetchTime, 3600) * time.Second
+	p.mu.Unlock()
+	ticker := time.NewTicker(refetchInterval)
 	defer ticker.Stop()
 
 	logger := mainLog.Load().With().Str("mode", "api-reload").Logger()
@@ -550,12 +693,18 @@ func (p *prog) apiConfigReload() {
 	}
 
 	doReloadApiConfig := func(forced bool, logger zerolog.Logger) {
+		if ctx.Err() != nil {
+			return
+		}
 		req := &controld.ResolverConfigRequest{
 			RawUID:   cdUID,
 			Version:  rootCmd.Version,
-			Metadata: ctrld.SystemMetadataRuntime(context.Background()),
+			Metadata: ctrld.SystemMetadataRuntime(ctx),
+		}
+		resolverConfig, err := fetchResolverConfigForReload(ctx, req, cdDev)
+		if ctx.Err() != nil {
+			return
 		}
-		resolverConfig, err := controld.FetchResolverConfig(context.Background(), req, cdDev)
 		selfUninstallCheck(err, p, logger)
 		if err != nil {
 			logger.Warn().Err(err).Msg("could not fetch resolver config")
@@ -608,7 +757,10 @@ func (p *prog) apiConfigReload() {
 			if !noInternalDomainsChanged {
 				logger.Info().Msg("internal domain changes detected, reloading...")
 			}
-			p.apiReloadCh <- nil
+			select {
+			case p.apiReloadCh <- nil:
+			case <-ctx.Done():
+			}
 			return
 		}
 
@@ -623,13 +775,16 @@ func (p *prog) apiConfigReload() {
 			}
 			if cfgErr != nil {
 				logger.Warn().Err(err).Msg("skipping invalid custom config")
-				if _, err := controld.UpdateCustomLastFailed(context.Background(), cdUID, rootCmd.Version, cdDev, true); err != nil {
+				if _, err := controld.UpdateCustomLastFailed(ctx, cdUID, rootCmd.Version, cdDev, true); err != nil {
 					logger.Error().Err(err).Msg("could not mark custom last update failed")
 				}
 				return
 			}
 			logger.Debug().Msg("custom config changes detected, reloading...")
-			p.apiReloadCh <- cfg
+			select {
+			case p.apiReloadCh <- cfg:
+			case <-ctx.Done():
+			}
 		} else {
 			logger.Debug().Msg("custom config does not change")
 		}
@@ -640,7 +795,7 @@ func (p *prog) apiConfigReload() {
 			doReloadApiConfig(true, logger.With().Bool("forced", true).Logger())
 		case <-ticker.C:
 			doReloadApiConfig(false, logger)
-		case <-p.stopCh:
+		case <-ctx.Done():
 			return
 		}
 	}
@@ -710,8 +865,21 @@ func (p *prog) setupUpstream(cfg *ctrld.Config) {
 // out the full timeout. A terminal provisioning path calls this right
 // before failing.
 func (p *prog) notifyExitToLogServer() {
-	if p.logConn != nil {
-		_, _ = p.logConn.Write([]byte(msgExit))
+	p.logConnMu.Lock()
+	conn := p.logConn
+	p.logConnMu.Unlock()
+	if conn != nil {
+		_, _ = conn.Write([]byte(msgExit))
+	}
+}
+
+func (p *prog) closeLogConn() {
+	p.logConnMu.Lock()
+	conn := p.logConn
+	p.logConn = nil
+	p.logConnMu.Unlock()
+	if conn != nil {
+		_ = conn.Close()
 	}
 }
 
@@ -734,7 +902,16 @@ func (p *prog) reportServeDNSFailure(listenerNum string, err error) {
 // so all listeners could be terminated and re-spawned again.
 func (p *prog) run(reload bool, reloadCh chan struct{}) {
 	// Wait the caller to signal that we can do our logic.
-	<-p.waitCh
+	select {
+	case <-p.waitCh:
+	case <-p.stopCh:
+		return
+	case <-p.runAbortCh:
+		return
+	}
+	if stopRequested(p.stopCh) || stopRequested(p.runAbortCh) {
+		return
+	}
 	if !reload {
 		p.preRun()
 	}
@@ -778,7 +955,6 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) {
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(len(p.cfg.Listener))
 
 	for _, nc := range p.cfg.Network {
 		for _, cidr := range nc.Cidrs {
@@ -809,7 +985,21 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) {
 
 	// context for managing spawn goroutines.
 	ctx, cancelFunc := context.WithCancel(context.Background())
-	defer cancelFunc()
+	defer func() {
+		cancelFunc()
+		wg.Wait()
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		select {
+		case <-p.stopCh:
+		case <-p.runAbortCh:
+		case <-reloadCh:
+		case <-ctx.Done():
+		}
+		cancelFunc()
+	}()
 
 	// Newer versions of android and iOS denies permission which breaks connectivity.
 	if !isMobile() && !reload {
@@ -822,13 +1012,19 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) {
 
 	if !reload {
 		p.newNetworkJournal()
-		go p.startNetworkMonitorAndJournal()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			p.waitNetworkJournal = p.startNetworkMonitorAndJournal()
+		}()
 	}
 
 	for listenerNum := range p.cfg.Listener {
 		p.cfg.Listener[listenerNum].Init()
 		if !reload {
+			p.listenerWg.Add(1)
 			go func(listenerNum string) {
+				defer p.listenerWg.Done()
 				listenerConfig := p.cfg.Listener[listenerNum]
 				upstreamConfig := p.cfg.Upstream[listenerNum]
 				if upstreamConfig == nil {
@@ -836,31 +1032,30 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) {
 				}
 				addr := net.JoinHostPort(listenerConfig.IP, strconv.Itoa(listenerConfig.Port))
 				mainLog.Load().Info().Msgf("starting DNS server on listener.%s: %s", listenerNum, addr)
-				if err := p.serveDNS(listenerNum); err != nil {
+				if err := serveDNSFn(p, listenerNum); err != nil {
 					p.reportServeDNSFailure(listenerNum, err)
 				}
 				mainLog.Load().Debug().Msgf("end of serveDNS listener.%s: %s", listenerNum, addr)
 			}(listenerNum)
 		}
-		go func() {
-			defer func() {
-				cancelFunc()
-				wg.Done()
-			}()
-			select {
-			case <-p.stopCh:
-			case <-ctx.Done():
-			case <-reloadCh:
-			}
-		}()
 	}
 
 	if !reload {
 		for i := 0; i < numListeners; i++ {
-			<-p.started
+			select {
+			case <-p.started:
+			case <-p.stopCh:
+				return
+			case <-p.runAbortCh:
+				return
+			}
 		}
-		for _, f := range p.onStarted {
-			f()
+		if !p.startOSState(func() {
+			for _, f := range p.onStarted {
+				f()
+			}
+		}) {
+			return
 		}
 	}
 
@@ -886,11 +1081,9 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) {
 		// Stop writing log to unix socket.
 		consoleWriter.Out = os.Stdout
 		p.initLogging(false)
-		if p.logConn != nil {
-			_ = p.logConn.Close()
-		}
-		go p.apiConfigReload()
-		p.postRun()
+		p.closeLogConn()
+		p.startAPIConfigReload()
+		p.startOSState(func() { postRunFn(p) })
 	}
 	wg.Wait()
 }
@@ -938,7 +1131,7 @@ func (p *prog) newNetworkJournal() {
 // journal. The trackers do not depend on the monitor, so a monitor that cannot
 // start leaves them running. It passes on the wait function of the journal.
 func (p *prog) startNetworkMonitorAndJournal() func() {
-	if err := p.monitorNetworkChanges(); err != nil {
+	if err := monitorNetworkChangesFn(p); err != nil {
 		mainLog.Load().Error().Err(err).Msg("Failed to start network monitoring")
 	}
 	return p.startNetworkJournal()
@@ -949,12 +1142,21 @@ func (p *prog) startNetworkMonitorAndJournal() func() {
 // table of each moment after an incident. It returns a function that waits for
 // both loops, so a caller that closed the stop channel knows when they ended.
 func (p *prog) startNetworkJournal() func() {
-	healthDone := p.health.startLoop(p.stopCh, func() (int, bool) {
+	// Journal trackers persist across reloads, but not a stop or startup abort.
+	stop := make(chan struct{})
+	go func() {
+		select {
+		case <-p.stopCh:
+		case <-p.runAbortCh:
+		}
+		close(stop)
+	}()
+	healthDone := p.health.startLoop(stop, func() (int, bool) {
 		return p.upstreamMonitorNow().countDownExcept(isInternalDomainUpstream), p.recoveryBypass.Load()
 	})
 	var configDone <-chan struct{}
 	if p.dnsConfig != nil {
-		configDone = p.dnsConfig.startLoop(p.stopCh, p.logDNSConfigChanges)
+		configDone = p.dnsConfig.startLoop(stop, p.logDNSConfigChanges)
 	}
 	p.logNetworkSnapshot("start")
 	return func() {
@@ -1012,22 +1214,124 @@ func (p *prog) metricsEnabled() bool {
 	return p.cfg.Service.MetricsQueryStats || p.cfg.Service.MetricsListener != ""
 }
 
-func (p *prog) Stop(s service.Service) error {
-	p.stopDnsWatchers()
-	mainLog.Load().Debug().Msg("dns watchers stopped")
-	for _, f := range p.onStopped {
-		f()
+// finishRun also cancels workers waiting for startup when preflight returns early.
+// The caller's stop channel belongs to the mobile controller, not to this cleanup.
+func (p *prog) finishRun() {
+	// Keep the files open through final diagnostics, including mobile exits
+	// that never invoke the service Stop callback.
+	defer p.closeInternalLogs()
+	close(p.runAbortCh)
+	select {
+	case <-p.runDone:
+	case <-time.After(shutdownTimeout):
+		mainLog.Load().Warn().Msg("timeout waiting for ctrld components to stop")
+		// A timeout is not proof that workers stopped. Never release resources
+		// still owned by the run, or allow a mobile restart to overlap it.
+		<-p.runDone
 	}
-	mainLog.Load().Debug().Msg("finish running onStopped functions")
+	// Listeners live across config reloads, so they belong to prog rather
+	// than the per-run wait group. runDone closes after their final Add.
+	p.apiReloadWg.Wait()
+	p.listenerWg.Wait()
+	if p.waitNetworkJournal != nil {
+		p.waitNetworkJournal()
+	}
+	if err := p.shutdown(); err != nil {
+		mainLog.Load().Warn().Err(err).Msg("error during shutdown")
+	}
+}
+
+// startOSState serializes startup's OS mutations with restoration. A stop
+// either restores an already-completed action or prevents it from starting.
+func (p *prog) startOSState(f func()) bool {
+	p.osStateMu.Lock()
+	defer p.osStateMu.Unlock()
+	if p.osStateRestored || stopRequested(p.stopCh) || stopRequested(p.runAbortCh) {
+		return false
+	}
+	f()
+	return true
+}
+
+// restoreOSState puts back what ctrld changed outside the process: the DNS
+// settings, the router configuration and the allocated listener IPs. It runs
+// while the listeners are still serving, so the OS is never left pointing at a
+// resolver that is already gone. Safe to call multiple times.
+func (p *prog) restoreOSState() error {
+	p.restoreOnce.Do(func() {
+		p.osStateMu.Lock()
+		p.osStateRestored = true
+		p.osStateMu.Unlock()
+		p.closeNetMonitor()
+		p.stopDnsWatchers()
+		mainLog.Load().Debug().Msg("dns watchers stopped")
+		for _, f := range p.onStopped {
+			f()
+		}
+		mainLog.Load().Debug().Msg("finish running onStopped functions")
+		if derr := p.deAllocateIP(); derr != nil {
+			mainLog.Load().Error().Err(derr).Msg("de-allocate ip failed")
+			p.restoreErr = derr
+		}
+	})
+	return p.restoreErr
+}
+
+// releaseResources releases the control server and upstream transports. It must run after
+// the listeners stopped, since retiring an upstream a listener still answers
+// queries on would fail those queries. Safe to call multiple times.
+func (p *prog) releaseResources() {
+	p.releaseOnce.Do(func() {
+		if p.cs != nil {
+			if cerr := p.cs.stop(); cerr != nil {
+				mainLog.Load().Warn().Err(cerr).Msg("could not stop control server")
+			}
+		}
+		p.closeLogConn()
+		p.mu.Lock()
+		upstreams := p.cfg.Upstream
+		p.mu.Unlock()
+		closeReplacedUpstreams(upstreams, nil)
+	})
+}
+
+// shutdown runs the full teardown, in the order both halves require. It is
+// called once the listeners stopped, on every stop path including the mobile
+// one where the OS never terminates the process.
+func (p *prog) shutdown() error {
+	err := p.restoreOSState()
+	p.releaseResources()
+	return err
+}
+
+// closeReplacedUpstreams releases the transports of upstreams that cur no longer
+// refers to, or all of them when cur is nil. Requests in flight on them fail
+// fast and are retried, the same way a re-bootstrap treats connections it
+// replaces.
+func closeReplacedUpstreams(old, cur map[string]*ctrld.UpstreamConfig) {
+	inUse := make(map[*ctrld.UpstreamConfig]struct{}, len(cur))
+	for _, uc := range cur {
+		inUse[uc] = struct{}{}
+	}
+	for _, uc := range old {
+		if uc == nil {
+			continue
+		}
+		if _, ok := inUse[uc]; !ok {
+			uc.CloseTransports()
+		}
+	}
+}
+
+func (p *prog) Stop(s service.Service) error {
 	defer func() {
 		mainLog.Load().Info().Msg("Service stopped")
-		// The files take the last line of this run before they close.
 		p.closeInternalLogs()
 	}()
-	if err := p.deAllocateIP(); err != nil {
-		mainLog.Load().Error().Err(err).Msg("de-allocate ip failed")
-		return err
-	}
+	// Only the OS level state is restored here. The listeners are still serving
+	// at this point, so releasing the upstreams they answer queries on has to
+	// wait until run observes stopCh and they have stopped.
+	err := p.restoreOSState()
 	if deactivationPinSet() {
 		select {
 		case <-p.pinCodeValidCh:
@@ -1052,7 +1356,7 @@ func (p *prog) Stop(s service.Service) error {
 		}
 	}
 	close(p.stopCh)
-	return nil
+	return err
 }
 
 func (p *prog) stopDnsWatchers() {
@@ -1080,7 +1384,7 @@ func (p *prog) deAllocateIP() error {
 		return nil
 	}
 	for _, lc := range p.cfg.Listener {
-		if err := deAllocateIP(lc.IP); err != nil {
+		if err := deAllocateIPFn(lc.IP); err != nil {
 			return err
 		}
 	}
@@ -1095,6 +1399,10 @@ func (p *prog) deAllocateIP() error {
 // NRPT rule, so a test of what happens *after* it fails must not be the thing that
 // runs it.
 var (
+	deAllocateIPFn                              = deAllocateIP
+	serveDNSFn                                  = (*prog).serveDNS
+	monitorNetworkChangesFn                     = (*prog).monitorNetworkChanges
+	postRunFn                                   = (*prog).postRun
 	localResolverIPFn                           = router.LocalResolverIP
 	startDNSInterceptFn                         = (*prog).startDNSIntercept
 	ensureInterceptDNSTargetFn                  = (*prog).ensureInterceptDNSTarget

cmd/cli/recovery_state.go

@@ -9,6 +9,12 @@ func (p *prog) beginRecovery(reason RecoveryReason) (ctx context.Context, gen ui
 	p.recoveryCancelMu.Lock()
 	defer p.recoveryCancelMu.Unlock()
 
+	// Close publishes the gate before taking recoveryCancelMu to cancel the
+	// current owner. Thus an admitted worker either publishes its cancel first
+	// or observes shutdown here; it cannot create an uncancellable late owner.
+	if p.networkActivityClosed() {
+		return nil, 0, false, false
+	}
 	if reason != RecoveryReasonNetworkChange && p.recoveryCancel != nil {
 		return nil, 0, false, false
 	}
@@ -28,7 +34,7 @@ func (p *prog) beginRecovery(reason RecoveryReason) (ctx context.Context, gen ui
 func (p *prog) recoveryOwnsState(gen uint64) bool {
 	p.recoveryCancelMu.Lock()
 	defer p.recoveryCancelMu.Unlock()
-	return p.recoveryGen.Load() == gen && p.recoveryCancel != nil
+	return !p.networkActivityClosed() && p.recoveryGen.Load() == gen && p.recoveryCancel != nil
 }
 
 func systemNameserversForInterceptRetry() []string {

cmd/cli/reload_others.go

@@ -12,6 +12,10 @@ func notifyReloadSigCh(ch chan os.Signal) {
 	signal.Notify(ch, syscall.SIGUSR1)
 }
 
+func stopNotifyReloadSigCh(ch chan os.Signal) {
+	signal.Stop(ch)
+}
+
 func (p *prog) sendReloadSignal() error {
 	return syscall.Kill(syscall.Getpid(), syscall.SIGUSR1)
 }

cmd/cli/reload_windows.go

@@ -8,6 +8,8 @@ import (
 
 func notifyReloadSigCh(ch chan os.Signal) {}
 
+func stopNotifyReloadSigCh(ch chan os.Signal) {}
+
 func (p *prog) sendReloadSignal() error {
 	select {
 	case p.reloadCh <- struct{}{}:

cmd/cli/restore_lock_test.go

@@ -0,0 +1,51 @@
+package cli
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+func TestRestoreOSStateReleasesMutexBeforeCallbackDrain(t *testing.T) {
+	p := &prog{cfg: &ctrld.Config{}, dnsWatcherStopCh: make(chan struct{})}
+	entered, resume, callbackDone := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	var startupCalls, restoreCalls atomic.Int32
+	p.onStopped = []func(){func() {
+		select {
+		case <-callbackDone:
+		default:
+			t.Error("OS restoration ran before the callback drained")
+		}
+		restoreCalls.Add(1)
+	}}
+	go p.networkChangeCallback(func(*netmon.ChangeDelta) {
+		close(entered)
+		<-resume
+		// This callback was admitted before shutdown. It must be able to take
+		// osStateMu and observe the closed startup gate while shutdown joins it.
+		if p.startOSState(func() { startupCalls.Add(1) }) {
+			t.Error("startup was permitted after restoration began")
+		}
+		close(callbackDone)
+	})(nil)
+	waitNetworkShutdownSignal(t, entered, "callback entry")
+	restored := make(chan struct{})
+	go func() {
+		if err := p.restoreOSState(); err != nil {
+			t.Error(err)
+		}
+		close(restored)
+	}()
+	waitNetworkShutdownGate(t, p)
+	close(resume)
+	waitNetworkShutdownSignal(t, callbackDone, "callback acquiring osStateMu")
+	waitNetworkShutdownSignal(t, restored, "OS restoration")
+	if err := p.restoreOSState(); err != nil {
+		t.Fatal(err)
+	}
+	if startupCalls.Load() != 0 || restoreCalls.Load() != 1 {
+		t.Fatalf("startup calls=%d, restore calls=%d", startupCalls.Load(), restoreCalls.Load())
+	}
+}

cmd/cli/shutdown_test.go

@@ -0,0 +1,164 @@
+package cli
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/Control-D-Inc/ctrld"
+	"github.com/Control-D-Inc/ctrld/internal/controld"
+	"github.com/Control-D-Inc/ctrld/internal/router"
+)
+
+func TestRunWaitStopsBeforeStartup(t *testing.T) {
+	p := &prog{
+		waitCh:  make(chan struct{}),
+		stopCh:  make(chan struct{}),
+		runDone: make(chan struct{}),
+	}
+	require.NoError(t, p.Start(nil))
+	close(p.stopCh)
+	select {
+	case <-p.runDone:
+	case <-time.After(time.Second):
+		t.Fatal("stop did not terminate workers waiting for startup")
+	}
+}
+
+func TestRunDoesNotStartAfterStop(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		p := &prog{waitCh: make(chan struct{}), stopCh: make(chan struct{})}
+		close(p.waitCh)
+		close(p.stopCh)
+		// cfg is deliberately nil: reaching component startup is a failure.
+		p.run(false, make(chan struct{}))
+	}
+}
+
+type startupTestRouter struct {
+	router.Router
+	preRunCalls int
+}
+
+func (r *startupTestRouter) PreRun() error {
+	r.preRunCalls++
+	return nil
+}
+
+func TestRunMobilePreflightCleanup(t *testing.T) {
+	const childEnv = "CTRLD_TEST_MOBILE_PREFLIGHT"
+	mode := os.Getenv(childEnv)
+	if mode == "" {
+		for _, mode := range []string{"stop", "failure"} {
+			t.Run(mode, func(t *testing.T) {
+				ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+				defer cancel()
+				cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestRunMobilePreflightCleanup$", "-test.v")
+				cmd.Env = append(os.Environ(), childEnv+"="+mode)
+				out, err := cmd.CombinedOutput()
+				require.NoError(t, err, "%s", out)
+			})
+		}
+		return
+	}
+
+	// Keep the real RunMobile -> run -> Start -> runWait path, but replace the
+	// platform/router and API boundaries before any host setup can happen.
+	isMobile = func() bool { return true }
+	networkUp = func() bool { return true }
+	r := &startupTestRouter{}
+	newRouter = func(*ctrld.Config, bool) router.Router { return r }
+	dir := t.TempDir()
+	configPath = filepath.Join(dir, "ctrld.toml")
+	require.NoError(t, os.WriteFile(configPath, []byte("[service]\n"), 0600))
+	want := errors.New("preflight rejected")
+	for i := 0; i < 3; i++ {
+		stopCh := make(chan struct{})
+		fetchCalls, exits := 0, 0
+		processCDFlagsFn = func(ctx context.Context, _ *ctrld.Config) (*controld.ResolverConfig, error) {
+			fetchCalls++
+			require.Eventually(t, startupWorkersRunning, time.Second, time.Millisecond)
+			if mode == "stop" {
+				close(stopCh)
+				<-ctx.Done()
+				return nil, ctx.Err()
+			}
+			return nil, want
+		}
+		callback := AppCallback{Exit: func(err string) {
+			exits++
+			require.Equal(t, want.Error(), err)
+		}}
+		RunMobile(&AppConfig{CdUID: "testuid", HomeDir: dir, UpstreamProto: "doh", LogPath: filepath.Join(dir, "ctrld.log")}, &callback, stopCh)
+		require.Equal(t, 1, fetchCalls)
+		if mode == "failure" {
+			require.Equal(t, 1, exits)
+			require.False(t, stopRequested(stopCh), "cleanup must not close the controller-owned channel")
+
+		} else {
+			require.Zero(t, exits)
+		}
+		require.False(t, startupWorkersRunning(), "RunMobile returned before its workers stopped")
+		if mode == "failure" {
+			close(stopCh) // A later Controller.Stop must remain safe.
+		}
+	}
+	require.Equal(t, 3, r.preRunCalls, "the real router must never be reached")
+}
+
+func startupWorkersRunning() bool {
+	stack := make([]byte, 1<<20)
+	n := runtime.Stack(stack, true)
+	return strings.Contains(string(stack[:n]), "(*prog).runWait(") || strings.Contains(string(stack[:n]), "(*prog).run(")
+}
+
+func TestControlServerStopOnlyUnlinksOwnedListener(t *testing.T) {
+	// Keep the Unix socket path below macOS's limit, including its long TMPDIR.
+	dir, err := os.MkdirTemp("", "ctrld-")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+	path := filepath.Join(dir, "control.sock")
+	owner, err := newControlServer(path)
+	require.NoError(t, err)
+	require.NoError(t, owner.start())
+	t.Cleanup(func() { _ = owner.stop() })
+	unstarted, err := newControlServer(path)
+	require.NoError(t, err)
+	require.NoError(t, unstarted.stop())
+	c, err := net.Dial("unix", path)
+	require.NoError(t, err, "stopping before startup must preserve someone else's socket")
+	require.NoError(t, c.SetDeadline(time.Now().Add(time.Second)))
+	_, err = io.WriteString(c, "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
+	require.NoError(t, err)
+	resp, err := http.ReadResponse(bufio.NewReader(c), nil)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	require.NoError(t, c.Close())
+	require.NoError(t, owner.stop())
+	_, err = os.Stat(path)
+	require.True(t, os.IsNotExist(err), "owned socket must be unlinked on shutdown: %v", err)
+}
+
+func TestShutdownClosesStartupLogConnection(t *testing.T) {
+	c, peer := net.Pipe()
+	defer peer.Close()
+	p := &prog{cfg: &ctrld.Config{}, dnsWatcherStopCh: make(chan struct{}), logConn: c}
+	require.NoError(t, p.shutdown())
+	require.NoError(t, p.shutdown())
+	_ = peer.SetWriteDeadline(time.Now().Add(time.Second))
+	_, err := peer.Write([]byte("x"))
+	require.ErrorIs(t, err, io.ErrClosedPipe)
+}

cmd/cli/upstream_retirement_test.go

@@ -0,0 +1,101 @@
+package cli
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"net"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/miekg/dns"
+	"github.com/stretchr/testify/require"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestCloseReplacedUpstreamsPreservesRetainedAndNew(t *testing.T) {
+	certServer := httptest.NewTLSServer(nil)
+	certificate := certServer.TLS.Certificates[0]
+	roots := x509.NewCertPool()
+	roots.AddCert(certServer.Certificate())
+	certServer.Close()
+
+	type upstream struct {
+		config  *ctrld.UpstreamConfig
+		entered <-chan struct{}
+		release func()
+	}
+	newUpstream := func() upstream {
+		ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{certificate}})
+		require.NoError(t, err)
+		entered, release := make(chan struct{}), make(chan struct{})
+		var enteredOnce, releaseOnce sync.Once
+		s := &dns.Server{Listener: ln, Net: "tcp-tls", Handler: dns.HandlerFunc(func(w dns.ResponseWriter, m *dns.Msg) {
+			enteredOnce.Do(func() { close(entered) })
+			<-release
+			answer := new(dns.Msg)
+			answer.SetReply(m)
+			_ = w.WriteMsg(answer)
+		})}
+		done := make(chan error, 1)
+		go func() { done <- s.ActivateAndServe() }()
+		uc := &ctrld.UpstreamConfig{Type: ctrld.ResolverTypeDOT, Endpoint: ln.Addr().String(), BootstrapIP: "127.0.0.1"}
+		uc.Init()
+		uc.SetCertPool(roots)
+		unblock := func() { releaseOnce.Do(func() { close(release) }) }
+		t.Cleanup(func() {
+			unblock()
+			uc.CloseTransports()
+			_ = s.Shutdown()
+			_ = ln.Close()
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Error("test DNS server did not stop")
+			}
+		})
+		return upstream{uc, entered, unblock}
+	}
+	removed, retained, replacement := newUpstream(), newUpstream(), newUpstream()
+	query := func(u upstream) <-chan error {
+		result := make(chan error, 1)
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			r, err := ctrld.NewResolver(u.config)
+			if err == nil {
+				q := new(dns.Msg)
+				q.SetQuestion("retirement.example.", dns.TypeA)
+				_, err = r.Resolve(ctx, q)
+			}
+			result <- err
+		}()
+		return result
+	}
+	removedResult, retainedResult := query(removed), query(retained)
+	for _, ch := range []<-chan struct{}{removed.entered, retained.entered} {
+		select {
+		case <-ch:
+		case <-time.After(3 * time.Second):
+			t.Fatal("query did not reach local upstream")
+		}
+	}
+	closeReplacedUpstreams(map[string]*ctrld.UpstreamConfig{
+		"0": removed.config, "1": retained.config, "nil": nil,
+	}, map[string]*ctrld.UpstreamConfig{
+		"0": replacement.config, "renamed": retained.config,
+	})
+	select {
+	case err := <-removedResult:
+		require.ErrorIs(t, err, net.ErrClosed)
+	case <-time.After(time.Second):
+		t.Fatal("removed upstream's active query was not closed")
+	}
+	retained.release()
+	require.NoError(t, <-retainedResult, "retained pointer must remain usable under a different key")
+	replacement.release()
+	require.NoError(t, <-query(replacement), "new upstream must not be retired with the old key")
+}