diff --git a/cmd/cli/api_reload_shutdown_test.go b/cmd/cli/api_reload_shutdown_test.go new file mode 100644 index 0000000..9864463 --- /dev/null +++ b/cmd/cli/api_reload_shutdown_test.go @@ -0,0 +1,160 @@ +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), + } + p.logger.Store(mainLog.Load()) + 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{}) + fetchResolverConfigFn = 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.logger.Store(mainLog.Load()) + 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} + p.logger.Store(mainLog.Load()) + 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 763b615..bf50741 100644 --- a/cmd/cli/cli.go +++ b/cmd/cli/cli.go @@ -48,6 +48,7 @@ const ( windowsForwardersFilename = ".forwarders.txt" oldBinSuffix = "_previous" msgExit = "$$EXIT$$" + shutdownTimeout = 10 * time.Second ) var ( @@ -179,10 +180,15 @@ func initCLI() *cobra.Command { } // 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 + cleanupStaleDNSInterceptStateFn = cleanupStaleDNSInterceptState +) + func updateConfigInterceptMode(cfg *ctrld.Config, mode string) bool { desired := "" switch mode { @@ -277,6 +283,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, } @@ -309,7 +317,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() { @@ -359,10 +375,10 @@ func run(appCallback *AppCallback, stopCh chan struct{}) { // outbound traffic), which would deny this process's own API bootstrap below and // leave it retrying forever - never reaching the cleanup that lives inside // intercept startup. No-op when no stale state exists. - cleanupStaleDNSInterceptState() + cleanupStaleDNSInterceptStateFn() // Wait for network up. - if !ctrldnet.Up() { + if !networkUp() { failRunUnclassified(p.Error(), "network is not up yet", p.notifyExitToLogServer) return } @@ -520,6 +536,9 @@ func run(appCallback *AppCallback, stopCh chan struct{}) { close(waitCh) <-stopCh + if !isMobile() { + p.finishRun() + } } // writeConfigFile writes the configuration to a file @@ -2286,8 +2305,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) @@ -2426,6 +2446,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 8bf12fc..e75e409 100644 --- a/cmd/cli/control_server.go +++ b/cmd/cli/control_server.go @@ -79,7 +79,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 95748b8..798c886 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/cmd/cli/dns_intercept_darwin.go @@ -1511,6 +1511,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) @@ -1534,6 +1538,9 @@ func (p *prog) pfStabilizationLoopWithMaxWait(ctx context.Context, stableRequire case <-pollTicker.C: } + if p.networkActivityClosed() { + return + } if p.pfExecBackoffActive() { continue } @@ -2105,6 +2112,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 } @@ -2113,7 +2124,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() @@ -2190,6 +2201,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 } @@ -2511,6 +2526,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. @@ -2534,7 +2553,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 6fb2b78..48db252 100644 --- a/cmd/cli/dns_intercept_windows.go +++ b/cmd/cli/dns_intercept_windows.go @@ -3069,6 +3069,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 3b6213e..f63bb4b 100644 --- a/cmd/cli/dns_proxy.go +++ b/cmd/cli/dns_proxy.go @@ -207,6 +207,7 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha defer s.Shutdown() select { case <-p.stopCh: + case <-p.runAbortCh: case <-gctx.Done(): case err := <-errCh: p.Warn().Err(err).Msg("Local IPv6 listener failed") @@ -227,6 +228,7 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha defer s.Shutdown() select { case <-p.stopCh: + case <-p.runAbortCh: case <-gctx.Done(): case err := <-errCh: p.Warn().Err(err).Msgf("Could not listen on %s: %s", proto, listenAddr) @@ -242,9 +244,18 @@ func (p *prog) startListeners(ctx context.Context, cfg *ctrld.ListenerConfig, ha addr := net.JoinHostPort(cfg.IP, strconv.Itoa(cfg.Port)) s, errCh := runDNSServer(addr, proto, handler) defer s.Shutdown() - p.started <- struct{}{} + select { + case p.started <- struct{}{}: + case <-p.stopCh: + return nil + case <-p.runAbortCh: + return nil + case <-gctx.Done(): + return nil + } select { case <-p.stopCh: + case <-p.runAbortCh: case <-gctx.Done(): case err := <-errCh: return err @@ -1479,7 +1490,7 @@ func runDNSServer(addr, network string, handler dns.Handler) (*dns.Server, <-cha startedCh := make(chan struct{}) s.NotifyStartedFunc = func() { sync.OnceFunc(func() { close(startedCh) })() } - errCh := make(chan error) + errCh := make(chan error, 1) go func() { defer close(errCh) if err := s.ListenAndServe(); err != nil { @@ -1642,6 +1653,10 @@ func (p *prog) selfUninstallCoolOfPeriod() { // forceFetchingAPI sends signal to force syncing API config if run in cd mode, // and the domain == "cdUID.verify.controld.com" func (p *prog) forceFetchingAPI(domain string) { + if !p.beginNetworkActivity() { + return + } + defer p.netMonitorWG.Done() if cdUID == "" { return } @@ -1658,9 +1673,32 @@ func (p *prog) forceFetchingAPI(domain string) { return } _ = p.apiForceReloadGroup.DoChan("force_sync_api", func() (interface{}, error) { - p.apiForceReloadCh <- struct{}{} + if !p.beginNetworkActivity() { + return nil, nil + } + defer p.netMonitorWG.Done() + shutdown := p.networkActivityDone() + select { + case <-shutdown: + return nil, nil + case p.apiForceReloadCh <- struct{}{}: + case <-p.stopCh: + return nil, nil + case <-p.runAbortCh: + return nil, nil + } // Wait here to prevent abusing API if we are flooded. - time.Sleep(timeDurationOrDefault(p.cfg.Service.ForceRefetchWaitTime, 30) * time.Second) + p.mu.Lock() + wait := timeDurationOrDefault(p.cfg.Service.ForceRefetchWaitTime, 30) * time.Second + p.mu.Unlock() + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-timer.C: + case <-shutdown: + case <-p.stopCh: + case <-p.runAbortCh: + } return nil, nil }) } @@ -1861,6 +1899,18 @@ func FlushDNSCache() error { return nil } +// networkChangeCallback fences even callbacks already dispatched by netmon +// when Close begins. The monitor does not track those callback goroutines. +func (p *prog) networkChangeCallback(fn netmon.ChangeFunc) netmon.ChangeFunc { + return func(delta *netmon.ChangeDelta) { + if !p.beginNetworkActivity() { + return + } + defer p.netMonitorWG.Done() + fn(delta) + } +} + // monitorNetworkChanges starts monitoring for network interface changes func (p *prog) monitorNetworkChanges(ctx context.Context) error { mon, err := newNetworkChangeMonitorFn(func(format string, args ...any) { @@ -1871,9 +1921,14 @@ func (p *prog) monitorNetworkChanges(ctx context.Context) error { return fmt.Errorf("creating network monitor: %w", err) } - mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) { + mon.RegisterChangeCallback(p.networkChangeCallback(func(delta *netmon.ChangeDelta) { p.handleNetworkChange(ctx, delta, mon.IsMajorChangeFrom(delta.Old, delta.New)) - }) + })) + if !p.setNetMonitor(mon) { + _ = mon.Close() + p.Debug().Msg("Network monitor discarded, ctrld is stopping") + return nil + } mon.Start() p.logger.Load().Debug().Msg("Network monitor started") @@ -2421,12 +2476,19 @@ func (p *prog) debounceRecovery(transitionID uint64) { } p.recoveryDebounceMu.Lock() defer p.recoveryDebounceMu.Unlock() + if p.networkActivityClosed() { + return + } if p.recoveryDebounceTimer != nil { p.recoveryDebounceTimer.Stop() p.Debug().Msg("Recovery debounce: resetting timer (rapid network change)") } p.recoveryDebounceTimer = time.AfterFunc(recoveryDebounceWindow, func() { + if !p.beginNetworkActivity() { + return + } + defer p.netMonitorWG.Done() p.networkSourceMu.Lock() if p.networkAcceptedGen.Load() != transitionID { p.networkSourceMu.Unlock() @@ -2488,6 +2550,10 @@ func (p *prog) handleRecovery(reason RecoveryReason) { } func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID uint64) { + if !p.beginNetworkActivity() { + return + } + defer p.netMonitorWG.Done() p.networkSourceMu.Lock() if transitionID != 0 && p.networkAcceptedGen.Load() != transitionID { p.networkSourceMu.Unlock() @@ -2508,6 +2574,14 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u targetBefore := p.interceptTargetSnapshot() outcome := recoveryOutcomeSuperseded defer func() { p.endRecovery(&diagnostic, outcome, targetBefore) }() + // Clean every exit before the final snapshot, including cancellation before mutation. + defer p.recoveryCanceledCleanup(gen) + if recoveryCtx.Err() != nil { + if p.recoveryGen.Load() == gen { + outcome = recoveryOutcomeCanceled + } + return + } if reason == RecoveryReasonNetworkChange { p.logger.Load().Debug().Msg("Network change recovery now owns shared recovery state") } @@ -2539,10 +2613,9 @@ func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID u upstreams := p.buildRecoveryUpstreams(reason) recovered, err := waitForUpstreamRecoveryFn(p, recoveryCtx, upstreams, &diagnostic) if err != nil { - if p.recoveryOwnsState(gen) { + if p.recoveryGen.Load() == gen { outcome = recoveryOutcomeCanceled } - p.recoveryCanceledCleanup(gen) return } diagnostic.recoveredUpstream = recovered @@ -2702,6 +2775,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string var wg sync.WaitGroup p.Debug().Msgf("Starting upstream recovery check for %d upstreams", len(upstreams)) + defer func() { cancel(); wg.Wait() }() for name, uc := range upstreams { wg.Add(1) @@ -2781,7 +2855,6 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string case <-ctx.Done(): return "", ctx.Err() } - wg.Wait() return recovered, nil } diff --git a/cmd/cli/force_reload_lifetime_test.go b/cmd/cli/force_reload_lifetime_test.go new file mode 100644 index 0000000..53d662c --- /dev/null +++ b/cmd/cli/force_reload_lifetime_test.go @@ -0,0 +1,49 @@ +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{})} + p.logger.Store(mainLog.Load()) + // 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..eea3ee0 --- /dev/null +++ b/cmd/cli/internal_log_shutdown_test.go @@ -0,0 +1,51 @@ +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(), + } + p.logger.Store(mainLog.Load()) + 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.Write([]byte("unexpected write\n")); !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..ae00283 --- /dev/null +++ b/cmd/cli/listener_shutdown_test.go @@ -0,0 +1,296 @@ +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), + } + p.logger.Store(mainLog.Load()) + // 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} + p.logger.Store(mainLog.Load()) + 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" + fetchResolverConfigFn = 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, context.Context) 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{}), + } + p.logger.Store(mainLog.Load()) + 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, ctx context.Context, n string) error { + p.sema = &shutdownSemaphore{entered: entered, resume: release} + return p.serveDNS(ctx, n) + } + } else { + serveDNSFn = func(p *prog, _ context.Context, _ 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 e9403ed..bf88880 100644 --- a/cmd/cli/network_monitor_test.go +++ b/cmd/cli/network_monitor_test.go @@ -15,6 +15,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..80b2aa6 --- /dev/null +++ b/cmd/cli/network_shutdown_test.go @@ -0,0 +1,332 @@ +package cli + +import ( + "context" + "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.logger.Store(mainLog.Load()) + 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{} + p.logger.Store(mainLog.Load()) + 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{}} + p.logger.Store(mainLog.Load()) + 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{}} + p.logger.Store(mainLog.Load()) + 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.logger.Store(mainLog.Load()) + 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{} + p.logger.Store(mainLog.Load()) + // 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{} + p.logger.Store(mainLog.Load()) + m := &shutdownTestMonitor{} + newNetworkChangeMonitorFn = func(func(string, ...any)) (networkChangeMonitor, error) { + if late { + p.closeNetMonitor() + } + return m, nil + } + if err := p.monitorNetworkChanges(context.Background()); 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.logger.Store(mainLog.Load()) + 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{})} + p.logger.Store(mainLog.Load()) + 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 2c9a8df..182ed98 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 037f64e..0079278 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -121,6 +121,7 @@ type prog struct { apiForceReloadCh chan struct{} apiForceReloadGroup singleflight.Group logConn io.WriteCloser + logConnMu sync.Mutex cs *controlServer logger atomic.Pointer[ctrld.Logger] csSetDnsDone chan struct{} @@ -361,6 +362,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() + p.Debug().Msg("network monitor stopped") + } + p.netMonitorWG.Wait() + }) } func (p *prog) Start(_ service.Service) error { @@ -370,10 +476,11 @@ func (p *prog) Start(_ 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 @@ -396,6 +503,11 @@ func (p *prog) runWait() { newCfg = apiCfg case <-p.stopCh: close(reloadCh) + <-done + return + case <-p.runAbortCh: + close(reloadCh) + <-done return } @@ -475,6 +587,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 @@ -490,6 +603,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() @@ -572,13 +686,41 @@ func (p *prog) postRun() { // calls. var fetchResolverConfigFn = 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 } - ticker := time.NewTicker(timeDurationOrDefault(p.cfg.Service.RefetchTime, 3600) * time.Second) + 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() + }() + + p.mu.Lock() + refetchInterval := timeDurationOrDefault(p.cfg.Service.RefetchTime, 3600) * time.Second + p.mu.Unlock() + ticker := time.NewTicker(refetchInterval) defer ticker.Stop() logger := p.logger.Load().With().Str("mode", "api-reload") @@ -596,13 +738,19 @@ func (p *prog) apiConfigReload() { } doReloadApiConfig := func(forced bool, logger *ctrld.Logger) { - loggerCtx := ctrld.LoggerCtx(context.Background(), p.logger.Load()) + if ctx.Err() != nil { + return + } + loggerCtx := ctrld.LoggerCtx(ctx, p.logger.Load()) req := &controld.ResolverConfigRequest{ RawUID: cdUID, Version: appVersion, Metadata: ctrld.SystemMetadata(loggerCtx), } resolverConfig, err := fetchResolverConfigFn(loggerCtx, req, cdDev) + if ctx.Err() != nil { + return + } selfUninstallCheck(err, p, logger) if err != nil { logger.Warn().Err(err).Msg("Could not fetch resolver config") @@ -622,7 +770,7 @@ func (p *prog) apiConfigReload() { doReloadApiConfig(true, logger.With().Bool("forced", true)) case <-ticker.C: doReloadApiConfig(false, logger) - case <-p.stopCh: + case <-ctx.Done(): return } } @@ -646,6 +794,9 @@ func (p *prog) applyFetchedResolverConfig( forced bool, lastUpdated int64, ) int64 { + if loggerCtx.Err() != nil { + return lastUpdated + } if resolverConfig.DeactivationPin != nil { newDeactivationPin := *resolverConfig.DeactivationPin curDeactivationPin := cdDeactivationPin.Load() @@ -693,7 +844,10 @@ func (p *prog) applyFetchedResolverConfig( logger.Info().Msg("Internal domain changes detected, reloading...") } p.firewallOnConfigReload() - p.apiReloadCh <- nil + select { + case p.apiReloadCh <- nil: + case <-loggerCtx.Done(): + } return lastUpdated } @@ -717,7 +871,10 @@ func (p *prog) applyFetchedResolverConfig( // Firewall mode: flush allowlist so DNS queries against the new // config repopulate it with IPs allowed under the updated policy. p.firewallOnConfigReload() - p.apiReloadCh <- cfg + select { + case p.apiReloadCh <- cfg: + case <-loggerCtx.Done(): + } } else { logger.Debug().Msg("Custom config does not change") } @@ -795,7 +952,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() } @@ -821,7 +987,19 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) { // context for managing spawned goroutines. Firewall mode needs it before // listeners start so its TTL reaper can run from the first DNS response. ctx, cancelFunc := context.WithCancel(context.Background()) - defer cancelFunc() + var wg sync.WaitGroup + 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() + }() if p.cfg.Service.CacheEnable { cacher, err := dnscache.NewLRUCache(p.cfg.Service.CacheSize) @@ -839,9 +1017,6 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) { // Synchronize firewall mode before listeners process DNS responses. p.syncFirewallMode(ctx) - var wg sync.WaitGroup - wg.Add(len(p.cfg.Listener)) - for _, nc := range p.cfg.Network { for _, cidr := range nc.Cidrs { _, ipNet, err := net.ParseCIDR(cidr) @@ -880,13 +1055,19 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) { if !reload { p.newNetworkJournal() - go p.startNetworkMonitorAndJournal(ctx) + wg.Add(1) + go func() { + defer wg.Done() + p.waitNetworkJournal = p.startNetworkMonitorAndJournal(ctx) + }() } 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 { @@ -897,31 +1078,30 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) { // serveCtx uses Background() context so listeners survive between reloads. // Changes to listeners config require a service restart, not just reload. serveCtx := context.Background() - if err := p.serveDNS(serveCtx, listenerNum); err != nil { + if err := serveDNSFn(p, serveCtx, listenerNum); err != nil { p.reportServeDNSFailure(listenerNum, err) } p.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 } } @@ -947,11 +1127,9 @@ func (p *prog) run(reload bool, reloadCh chan struct{}) { // Stop writing log to unix socket. consoleWriter = newHumanReadableZapCore(os.Stdout, consoleWriterLevel) 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() } @@ -999,7 +1177,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(ctx context.Context) func() { - if err := p.monitorNetworkChanges(ctx); err != nil { + if err := monitorNetworkChangesFn(p, ctx); err != nil { p.Error().Err(err).Msg("Failed to start network monitoring") } return p.startNetworkJournal() @@ -1010,12 +1188,21 @@ func (p *prog) startNetworkMonitorAndJournal(ctx context.Context) 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() { @@ -1068,22 +1255,123 @@ func (p *prog) metricsEnabled() bool { return p.cfg.Service.MetricsQueryStats || p.cfg.Service.MetricsListener != "" } -func (p *prog) Stop(_ service.Service) error { - p.stopDnsWatchers() - p.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): + p.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 } - p.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 { + p.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, interception state and 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() + p.Debug().Msg("dns watchers stopped") + for _, f := range p.onStopped { + f() + } + p.Debug().Msg("finish running onStopped functions") + if derr := p.deAllocateIP(); derr != nil { + p.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 { + p.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(_ service.Service) error { defer func() { p.Info().Msg("Service stopped") // The files take the last line of this run before they close. p.closeInternalLogs() }() - if err := p.deAllocateIP(); err != nil { - p.Error().Err(err).Msg("De-allocate ip failed") - return err - } + // Restore OS state while listeners still serve; retire resources after their drain. + err := p.restoreOSState() if deactivationPinSet() { select { case <-p.pinCodeValidCh: @@ -1108,7 +1396,7 @@ func (p *prog) Stop(_ service.Service) error { } } close(p.stopCh) - return nil + return err } func (p *prog) stopDnsWatchers() { @@ -1136,7 +1424,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 } } @@ -1151,6 +1439,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 startDNSInterceptFn = (*prog).startDNSIntercept ensureInterceptDNSTargetFn = (*prog).ensureInterceptDNSTarget removeInterceptDNSTargetFn = (*prog).removeInterceptDNSTarget diff --git a/cmd/cli/prog_log.go b/cmd/cli/prog_log.go index 456d635..5d8c9db 100644 --- a/cmd/cli/prog_log.go +++ b/cmd/cli/prog_log.go @@ -36,7 +36,15 @@ func (p *prog) Notice() *ctrld.LogEvent { // so a waiting "ctrld start" is not left waiting on it after this process // exits. A nil connection (log server never started) is a no-op. func (p *prog) notifyExitToLogServer() { - if p.logConn != nil { - _ = p.logConn.Close() + p.closeLogConn() +} + +func (p *prog) closeLogConn() { + p.logConnMu.Lock() + conn := p.logConn + p.logConn = nil + p.logConnMu.Unlock() + if conn != nil { + _ = conn.Close() } } diff --git a/cmd/cli/recovery_state.go b/cmd/cli/recovery_state.go index 2d3b66e..e3ff638 100644 --- a/cmd/cli/recovery_state.go +++ b/cmd/cli/recovery_state.go @@ -13,6 +13,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 } @@ -32,7 +38,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 (p *prog) systemNameserversForInterceptRetry() []string { diff --git a/cmd/cli/reload_others.go b/cmd/cli/reload_others.go index cf374a0..480e773 100644 --- a/cmd/cli/reload_others.go +++ b/cmd/cli/reload_others.go @@ -13,6 +13,8 @@ func notifyReloadSigCh(ch chan os.Signal) { signal.Notify(ch, syscall.SIGUSR1) } +func stopNotifyReloadSigCh(ch chan os.Signal) { signal.Stop(ch) } + // sendReloadSignal sends a reload signal to the current process 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 b60f796..3a8a8ac 100644 --- a/cmd/cli/reload_windows.go +++ b/cmd/cli/reload_windows.go @@ -9,6 +9,8 @@ import ( // notifyReloadSigCh is a no-op on Windows platforms func notifyReloadSigCh(ch chan os.Signal) {} +func stopNotifyReloadSigCh(ch chan os.Signal) {} + // sendReloadSignal sends a reload signal to the program func (p *prog) sendReloadSignal() error { select { diff --git a/cmd/cli/restore_lock_test.go b/cmd/cli/restore_lock_test.go new file mode 100644 index 0000000..75850b4 --- /dev/null +++ b/cmd/cli/restore_lock_test.go @@ -0,0 +1,52 @@ +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{})} + p.logger.Store(mainLog.Load()) + 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..b735648 --- /dev/null +++ b/cmd/cli/shutdown_test.go @@ -0,0 +1,156 @@ +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" +) + +func TestRunWaitStopsBeforeStartup(t *testing.T) { + p := &prog{ + waitCh: make(chan struct{}), + stopCh: make(chan struct{}), + runDone: make(chan struct{}), + } + p.logger.Store(mainLog.Load()) + 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{})} + p.logger.Store(mainLog.Load()) + close(p.waitCh) + close(p.stopCh) + // cfg is deliberately nil: reaching component startup is a failure. + p.run(false, make(chan struct{})) + } +} + +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 cleanup and API boundaries before any host setup can happen. + isMobile = func() bool { return true } + networkUp = func() bool { return true } + cleanupCalls := 0 + cleanupStaleDNSInterceptStateFn = func() { cleanupCalls++ } + 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, cleanupCalls, "the real stale intercept cleanup 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} + p.logger.Store(mainLog.Load()) + 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..f6a712a --- /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(context.Background()) + 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(ctx, 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") +}