fix: back off macOS pf watchdog exec storms

cmd/cli/dns_intercept_darwin.go

@@ -41,6 +41,11 @@ const (
 	// 2s re-check misses.
 	pfAnchorRecheckDelayLong = 4 * time.Second
 
+	// pfExecFailureBackoff suppresses repeated pfctl/scutil checks briefly after
+	// macOS reports local resource exhaustion. Without this, network-change storms
+	// can turn a pf ruleset race into a fork/file-descriptor exhaustion loop.
+	pfExecFailureBackoff = 5 * time.Second
+
 	// pfVPNInterfacePrefixes lists interface name prefixes that indicate VPN/tunnel
 	// interfaces on macOS. Used to add interface-specific DNS intercept rules so that
 	// VPN software with "pass out quick on <iface>" rules cannot bypass our intercept.
@@ -1244,6 +1249,15 @@ func (p *prog) ensurePFAnchorActive() bool {
 	if p.dnsInterceptState == nil {
 		return false
 	}
+	if !p.pfEnsureRunning.CompareAndSwap(false, true) {
+		mainLog.Load().Debug().Msg("DNS intercept watchdog: check already running, skipping duplicate")
+		return false
+	}
+	defer p.pfEnsureRunning.Store(false)
+
+	if p.pfExecBackoffActive() {
+		return false
+	}
 
 	// While stabilizing (VPN connecting), suppress all restores.
 	// The stabilization loop will restore once pf settles.
@@ -1277,6 +1291,9 @@ func (p *prog) ensurePFAnchorActive() bool {
 	// Check 1: anchor references in the main ruleset.
 	natOut, err := exec.Command("pfctl", "-sn").CombinedOutput()
 	if err != nil {
+		if p.pfBackoffResourceExhaustion(err, natOut, "dump NAT rules") {
+			return false
+		}
 		mainLog.Load().Warn().Err(err).Msg("DNS intercept watchdog: could not dump NAT rules")
 		return false
 	}
@@ -1288,6 +1305,9 @@ func (p *prog) ensurePFAnchorActive() bool {
 	if !needsRestore {
 		filterOut, err := exec.Command("pfctl", "-sr").CombinedOutput()
 		if err != nil {
+			if p.pfBackoffResourceExhaustion(err, filterOut, "dump filter rules") {
+				return false
+			}
 			mainLog.Load().Warn().Err(err).Msg("DNS intercept watchdog: could not dump filter rules")
 			return false
 		}
@@ -1305,6 +1325,9 @@ func (p *prog) ensurePFAnchorActive() bool {
 	if !needsRestore {
 		anchorFilter, err := exec.Command("pfctl", "-a", pfAnchorName, "-sr").CombinedOutput()
 		if err != nil || len(strings.TrimSpace(string(anchorFilter))) == 0 {
+			if p.pfBackoffResourceExhaustion(err, anchorFilter, "dump anchor filter rules") {
+				return false
+			}
 			mainLog.Load().Warn().Msg("DNS intercept watchdog: anchor has no filter rules — content was flushed")
 			needsRestore = true
 		}
@@ -1312,6 +1335,9 @@ func (p *prog) ensurePFAnchorActive() bool {
 	if !needsRestore {
 		anchorNat, err := exec.Command("pfctl", "-a", pfAnchorName, "-sn").CombinedOutput()
 		if err != nil || len(strings.TrimSpace(string(anchorNat))) == 0 {
+			if p.pfBackoffResourceExhaustion(err, anchorNat, "dump anchor NAT rules") {
+				return false
+			}
 			mainLog.Load().Warn().Msg("DNS intercept watchdog: anchor has no rdr rules — translation was flushed (will cause packet loop on lo0)")
 			needsRestore = true
 		}
@@ -1345,6 +1371,7 @@ func (p *prog) ensurePFAnchorActive() bool {
 	// Restore: re-inject anchor references into the main ruleset.
 	mainLog.Load().Info().Msg("DNS intercept watchdog: restoring pf anchor references")
 	if err := p.ensurePFAnchorReference(); err != nil {
+		p.pfBackoffResourceExhaustion(err, nil, "restore anchor references")
 		mainLog.Load().Error().Err(err).Msg("DNS intercept watchdog: failed to restore anchor references")
 		return true
 	}
@@ -1361,6 +1388,7 @@ func (p *prog) ensurePFAnchorActive() bool {
 	if err := os.WriteFile(pfAnchorFile, []byte(rulesStr), 0644); err != nil {
 		mainLog.Load().Error().Err(err).Msg("DNS intercept watchdog: failed to write anchor file")
 	} else if out, err := exec.Command("pfctl", "-a", pfAnchorName, "-f", pfAnchorFile).CombinedOutput(); err != nil {
+		p.pfBackoffResourceExhaustion(err, out, "load rebuilt anchor")
 		mainLog.Load().Error().Err(err).Msgf("DNS intercept watchdog: failed to load rebuilt anchor (output: %s)", strings.TrimSpace(string(out)))
 	} else {
 		flushPFStates()
@@ -1389,6 +1417,58 @@ func (p *prog) ensurePFAnchorActive() bool {
 	return true
 }
 
+func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Duration) {
+	time.AfterFunc(delay, func() {
+		if p.dnsInterceptState == nil {
+			return
+		}
+		p.refreshDNSAfterVPNSettle(reason)
+	})
+}
+
+func (p *prog) pfExecBackoffActive() bool {
+	until := p.pfExecBackoffUntil.Load()
+	if until == 0 {
+		return false
+	}
+	remaining := time.Until(time.UnixMilli(until))
+	if remaining <= 0 {
+		p.pfExecBackoffUntil.CompareAndSwap(until, 0)
+		return false
+	}
+	mainLog.Load().Debug().Dur("remaining", remaining).Msg("DNS intercept watchdog: suppressed during pf exec backoff")
+	return true
+}
+
+func (p *prog) pfBackoffResourceExhaustion(err error, output []byte, operation string) bool {
+	if !isResourceExhaustion(err, output) {
+		return false
+	}
+	until := time.Now().Add(pfExecFailureBackoff)
+	p.pfExecBackoffUntil.Store(until.UnixMilli())
+	mainLog.Load().Warn().Err(err).Dur("backoff", pfExecFailureBackoff).Str("operation", operation).
+		Msg("DNS intercept watchdog: backing off after local exec resource exhaustion")
+	return true
+}
+
+func isResourceExhaustion(err error, output []byte) bool {
+	if err == nil && len(output) == 0 {
+		return false
+	}
+	var msg string
+	if err != nil {
+		msg = err.Error()
+	}
+	if len(output) > 0 {
+		msg += "\n" + string(output)
+	}
+	msg = strings.ToLower(msg)
+	return strings.Contains(msg, "resource temporarily unavailable") ||
+		strings.Contains(msg, "too many open files") ||
+		strings.Contains(msg, "too many processes") ||
+		strings.Contains(msg, "cannot allocate memory")
+}
+
 // pfWatchdog periodically checks that our pf anchor is still active.
 // Other programs (e.g., Windscribe desktop app, macOS configd) can replace
 // scheduleDelayedRechecks schedules delayed re-checks after a network change event.
@@ -1401,8 +1481,19 @@ func (p *prog) ensurePFAnchorActive() bool {
 //
 // Two delays (2s and 4s) cover both fast and slow VPN teardowns.
 func (p *prog) scheduleDelayedRechecks() {
+	p.pfDelayedRecheckMu.Lock()
+	defer p.pfDelayedRecheckMu.Unlock()
+
+	for _, timer := range p.pfDelayedRecheckTimers {
+		if timer != nil {
+			timer.Stop()
+		}
+	}
+	p.pfDelayedRecheckTimers = p.pfDelayedRecheckTimers[:0]
+
 	for _, delay := range []time.Duration{pfAnchorRecheckDelay, pfAnchorRecheckDelayLong} {
-		time.AfterFunc(delay, func() {
+		delay := delay
+		timer := time.AfterFunc(delay, func() {
 			if p.dnsInterceptState == nil || p.pfStabilizing.Load() {
 				return
 			}
@@ -1417,6 +1508,7 @@ func (p *prog) scheduleDelayedRechecks() {
 				p.vpnDNS.Refresh(ctx, true)
 			}
 		})
+		p.pfDelayedRecheckTimers = append(p.pfDelayedRecheckTimers, timer)
 	}
 }
 

cmd/cli/dns_intercept_darwin_test.go

@@ -3,6 +3,7 @@
 package cli
 
 import (
+	"errors"
 	"strings"
 	"testing"
 
@@ -141,3 +142,47 @@ func TestPFAddressFamily(t *testing.T) {
 		}
 	}
 }
+
+func TestIsResourceExhaustion(t *testing.T) {
+	tests := []struct {
+		name   string
+		err    error
+		output []byte
+		want   bool
+	}{
+		{
+			name: "exec start failure",
+			err:  errors.New("fork/exec /sbin/pfctl: resource temporarily unavailable"),
+			want: true,
+		},
+		{
+			name:   "fd exhaustion from stderr output",
+			err:    errors.New("exit status 1"),
+			output: []byte("pfctl: Pipe: Too many open files"),
+			want:   true,
+		},
+		{
+			name: "process exhaustion from wrapped restore error",
+			err:  errors.New("failed to dump running filter rules: exit status 1 (output: too many processes)"),
+			want: true,
+		},
+		{
+			name:   "ordinary pf syntax failure",
+			err:    errors.New("exit status 1"),
+			output: []byte("pfctl: syntax error"),
+			want:   false,
+		},
+		{
+			name: "nil error and empty output",
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isResourceExhaustion(tt.err, tt.output); got != tt.want {
+				t.Fatalf("isResourceExhaustion() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

cmd/cli/dns_proxy.go

@@ -2209,6 +2209,9 @@ func (p *prog) completeRecovery(reason RecoveryReason, recovered string) error {
 // waitForUpstreamRecovery checks the provided upstreams concurrently until one recovers.
 // It returns the name of the recovered upstream or an error if the check times out.
 func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string]*ctrld.UpstreamConfig) (string, error) {
+	recoveryCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	recoveredCh := make(chan string, 1)
 	var wg sync.WaitGroup
 
@@ -2222,7 +2225,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 			attempts := 0
 			for {
 				select {
-				case <-ctx.Done():
+				case <-recoveryCtx.Done():
 					p.Debug().Msgf("Context canceled for upstream %s", name)
 					return
 				default:
@@ -2234,13 +2237,16 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 						select {
 						case recoveredCh <- name:
 							p.Debug().Msgf("Sent recovery notification for upstream %s", name)
+							cancel()
 						default:
 							p.Debug().Msg("Recovery channel full, another upstream already recovered")
 						}
 						return
 					}
 					p.Debug().Msgf("Upstream %s check failed, sleeping before retry", name)
-					time.Sleep(checkUpstreamBackoffSleep)
+					if !sleepWithContext(recoveryCtx, checkUpstreamBackoffSleep) {
+						return
+					}
 
 					// if this is the upstreamOS and it's the 3rd attempt (or multiple of 3),
 					// we should try to reinit the OS resolver to ensure we can recover
@@ -2268,6 +2274,17 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 	return recovered, nil
 }
 
+func sleepWithContext(ctx context.Context, d time.Duration) bool {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
 // buildRecoveryUpstreams constructs the map of upstream configurations to test.
 // For OS failures we supply the manual OS resolver upstream configuration.
 // For network change or regular failure we use the upstreams defined in p.cfg (ignoring OS).

cmd/cli/prog.go

@@ -177,6 +177,21 @@ type prog struct {
 	// interception with exponential backoff and auto-heals if broken.
 	pfMonitorRunning atomic.Bool //lint:ignore U1000 used on darwin
 
+	// pfEnsureRunning ensures only one pf anchor validation/restoration runs at a time.
+	// Network-change callbacks, delayed rechecks, and the periodic watchdog can all
+	// converge during macOS interface churn; concurrent pfctl/scutil exec storms can
+	// exhaust process/file limits and make the outage worse.
+	pfEnsureRunning atomic.Bool //lint:ignore U1000 used on darwin
+
+	// pfExecBackoffUntil suppresses pf anchor validation after pfctl/scutil execs
+	// fail due host resource exhaustion (fork unavailable, too many open files).
+	pfExecBackoffUntil atomic.Int64 //lint:ignore U1000 used on darwin
+
+	// pfDelayedRecheckTimers coalesces delayed DNS-intercept rechecks after noisy
+	// network changes. Protected by pfDelayedRecheckMu.
+	pfDelayedRecheckMu     sync.Mutex    //lint:ignore U1000 used on darwin
+	pfDelayedRecheckTimers []*time.Timer //lint:ignore U1000 used on darwin
+
 	// pfProbeExpected holds the domain name of a pending pf interception probe.
 	pfProbeExpected atomic.Value // string
 
@@ -1323,6 +1338,7 @@ func errNetworkError(err error) bool {
 		case errors.Is(opErr.Err, syscall.ECONNREFUSED),
 			errors.Is(opErr.Err, syscall.EINVAL),
 			errors.Is(opErr.Err, syscall.ENETUNREACH),
+			errors.Is(opErr.Err, syscall.EHOSTUNREACH),
 			errors.Is(opErr.Err, windowsENETUNREACH),
 			errors.Is(opErr.Err, windowsEINVAL),
 			errors.Is(opErr.Err, windowsECONNREFUSED),

cmd/cli/prog_test.go

@@ -1,6 +1,10 @@
 package cli
 
 import (
+	"context"
+	"net"
+	"net/url"
+	"syscall"
 	"testing"
 	"time"
 
@@ -11,6 +15,23 @@ import (
 	"github.com/Control-D-Inc/ctrld"
 )
 
+func TestErrNetworkErrorTreatsNoRouteAsNetworkError(t *testing.T) {
+	err := &net.OpError{Op: "dial", Net: "tcp", Err: syscall.EHOSTUNREACH}
+	assert.True(t, errNetworkError(err))
+	assert.True(t, errUrlNetworkError(&url.Error{Op: "Get", URL: "https://dns.controld.com", Err: err}))
+}
+
+func TestSleepWithContext(t *testing.T) {
+	assert.True(t, sleepWithContext(context.Background(), time.Millisecond))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	start := time.Now()
+	assert.False(t, sleepWithContext(ctx, time.Minute))
+	assert.Less(t, time.Since(start), 100*time.Millisecond)
+}
+
 func Test_prog_dnsWatchdogEnabled(t *testing.T) {
 	p := &prog{cfg: &ctrld.Config{}}
 

cmd/cli/vpn_dns.go

@@ -55,6 +55,9 @@ type vpnDNSManager struct {
 	// discoverVPNDNS is injected for tests so Refresh does not depend on the
 	// runner host's real VPN/virtual adapter state.
 	discoverVPNDNS func(context.Context) []ctrld.VPNDNSConfig
+	// refreshRunning keeps noisy network-change storms from running overlapping
+	// scutil/networksetup VPN DNS discovery work.
+	refreshRunning atomic.Bool
 	// Called when VPN DNS server list changes, to update intercept exemptions.
 	onServersChanged vpnDNSExemptFunc
 }
@@ -76,6 +79,11 @@ func newVPNDNSManager(logger *atomic.Pointer[ctrld.Logger], exemptFunc vpnDNSExe
 func (m *vpnDNSManager) Refresh(ctx context.Context, guardAgainstNoNameservers ...bool) {
 	logger := ctrld.LoggerFromCtx(ctx)
 	guardedRefresh := len(guardAgainstNoNameservers) > 0 && guardAgainstNoNameservers[0]
+	if !m.refreshRunning.CompareAndSwap(false, true) {
+		ctrld.Log(ctx, logger.Debug(), "VPN DNS refresh already running, skipping duplicate")
+		return
+	}
+	defer m.refreshRunning.Store(false)
 
 	ctrld.Log(ctx, logger.Debug(), "Refreshing VPN DNS configurations")
 	discoverVPNDNS := m.discoverVPNDNS

cmd/cli/vpn_dns_test.go

@@ -2,6 +2,8 @@ package cli
 
 import (
 	"context"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/Control-D-Inc/ctrld"
@@ -14,6 +16,36 @@ func withVPNDNSSettlingEnabled(t *testing.T) {
 	t.Cleanup(func() { vpnDNSSettlingEnabled = old })
 }
 
+func TestVPNDNSRefreshSkipsConcurrentDuplicate(t *testing.T) {
+	m := newVPNDNSManager(nil)
+	started := make(chan struct{})
+	release := make(chan struct{})
+	done := make(chan struct{})
+	var once sync.Once
+	var calls atomic.Int32
+
+	m.discoverVPNDNS = func(context.Context) []ctrld.VPNDNSConfig {
+		calls.Add(1)
+		once.Do(func() { close(started) })
+		<-release
+		return nil
+	}
+
+	go func() {
+		defer close(done)
+		m.Refresh(true)
+	}()
+
+	<-started
+	m.Refresh(true)
+	close(release)
+	<-done
+
+	if calls.Load() != 1 {
+		t.Fatalf("expected overlapping refresh to be skipped, got %d discovery calls", calls.Load())
+	}
+}
+
 func TestVPNDNSRefreshRetainsStateForOneGuardedEmptyDiscovery(t *testing.T) {
 	withVPNDNSSettlingEnabled(t)
 	var gotExemptions []vpnDNSExemption