diff --git a/cmd/cli/api_reload_shutdown_test.go b/cmd/cli/api_reload_shutdown_test.go new file mode 100644 index 0000000..c52e6d5 --- /dev/null +++ b/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) +} diff --git a/cmd/cli/cli.go b/cmd/cli/cli.go index 7e86915..8679f40 100644 --- a/cmd/cli/cli.go +++ b/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()) diff --git a/cmd/cli/control_server.go b/cmd/cli/control_server.go index c6ba7f9..3e544de 100644 --- a/cmd/cli/control_server.go +++ b/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) diff --git a/cmd/cli/dns_intercept_darwin.go b/cmd/cli/dns_intercept_darwin.go index 7336b84..2402f3e 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/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 } diff --git a/cmd/cli/dns_intercept_windows.go b/cmd/cli/dns_intercept_windows.go index 4dd87d2..ce96536 100644 --- a/cmd/cli/dns_intercept_windows.go +++ b/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 } diff --git a/cmd/cli/dns_proxy.go b/cmd/cli/dns_proxy.go index bb6b0f5..4d3ca6f 100644 --- a/cmd/cli/dns_proxy.go +++ b/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 } diff --git a/cmd/cli/force_reload_lifetime_test.go b/cmd/cli/force_reload_lifetime_test.go new file mode 100644 index 0000000..b2600d3 --- /dev/null +++ b/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: + } +} diff --git a/cmd/cli/internal_log_shutdown_test.go b/cmd/cli/internal_log_shutdown_test.go new file mode 100644 index 0000000..5616105 --- /dev/null +++ b/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) + } + } +} diff --git a/cmd/cli/listener_shutdown_test.go b/cmd/cli/listener_shutdown_test.go new file mode 100644 index 0000000..b4f0bd7 --- /dev/null +++ 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() {} diff --git a/cmd/cli/network_monitor_test.go b/cmd/cli/network_monitor_test.go index 33dba8d..d23a9c1 100644 --- a/cmd/cli/network_monitor_test.go +++ b/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() {} diff --git a/cmd/cli/network_shutdown_test.go b/cmd/cli/network_shutdown_test.go new file mode 100644 index 0000000..714247a --- /dev/null +++ b/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") +} diff --git a/cmd/cli/network_transition.go b/cmd/cli/network_transition.go index 711e592..fb7825d 100644 --- a/cmd/cli/network_transition.go +++ b/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() diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index 8cf4a2d..b77d6c6 100644 --- a/cmd/cli/prog.go +++ b/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 diff --git a/cmd/cli/recovery_state.go b/cmd/cli/recovery_state.go index 0eae9cd..090e7e7 100644 --- a/cmd/cli/recovery_state.go +++ b/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 { diff --git a/cmd/cli/reload_others.go b/cmd/cli/reload_others.go index 0977af9..d72eab4 100644 --- a/cmd/cli/reload_others.go +++ b/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) } diff --git a/cmd/cli/reload_windows.go b/cmd/cli/reload_windows.go index 0e817e4..8d0054e 100644 --- a/cmd/cli/reload_windows.go +++ b/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{}{}: diff --git a/cmd/cli/restore_lock_test.go b/cmd/cli/restore_lock_test.go new file mode 100644 index 0000000..e80672d --- /dev/null +++ b/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()) + } +} diff --git a/cmd/cli/shutdown_test.go b/cmd/cli/shutdown_test.go new file mode 100644 index 0000000..eae8ae9 --- /dev/null +++ b/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) +} diff --git a/cmd/cli/upstream_retirement_test.go b/cmd/cli/upstream_retirement_test.go new file mode 100644 index 0000000..354fc8b --- /dev/null +++ b/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") +}