diff --git a/cmd/cli/dns_intercept_darwin.go b/cmd/cli/dns_intercept_darwin.go index 7b72df9c..364a29ad 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/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 " rules cannot bypass our intercept. @@ -1303,6 +1308,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. @@ -1336,6 +1350,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 } @@ -1348,6 +1365,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 } @@ -1365,6 +1385,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 } @@ -1372,6 +1395,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 } @@ -1405,6 +1431,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 } @@ -1421,6 +1448,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() @@ -1458,6 +1486,49 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati }) } +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. @@ -1470,8 +1541,19 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati // // 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 } @@ -1485,6 +1567,7 @@ func (p *prog) scheduleDelayedRechecks() { p.vpnDNS.Refresh(true) } }) + p.pfDelayedRecheckTimers = append(p.pfDelayedRecheckTimers, timer) } } diff --git a/cmd/cli/dns_intercept_darwin_test.go b/cmd/cli/dns_intercept_darwin_test.go index d0834d7f..c8d90b03 100644 --- a/cmd/cli/dns_intercept_darwin_test.go +++ b/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) + } + }) + } +} diff --git a/cmd/cli/dns_intercept_windows.go b/cmd/cli/dns_intercept_windows.go index a79fa0f6..5c3ecb34 100644 --- a/cmd/cli/dns_intercept_windows.go +++ b/cmd/cli/dns_intercept_windows.go @@ -278,6 +278,9 @@ type wfpState struct { loopbackProtectActive bool // loopbackPermitIDs stores the filter IDs for the loopback protect permits. loopbackPermitIDs []uint64 + // nrptRecoveryLimiter prevents repeated Windows policy/Dnscache signaling + // when another agent keeps putting NRPT back into a broken state. + nrptRecoveryLimiter nrptRecoveryLimiter } // Lazy-loaded WFP DLL procedures. @@ -343,9 +346,10 @@ const ( // - GP path: SOFTWARE\Policies\...\DnsPolicyConfig (Group Policy) // - Local path: SYSTEM\CurrentControlSet\...\DnsPolicyConfig (service store) // -// If ANY rules exist in the GP path (from IT policy, VPN, MDM, etc.), DNS Client -// enters "GP mode" and ignores ALL local-path rules entirely. Conversely, if the -// GP path is empty/absent, DNS Client reads from the local path only. +// If the GP path contains real rules (from IT policy, VPN, MDM, etc.), DNS +// Client enters "GP mode" and ignores ALL local-path rules entirely. An empty GP +// parent key is worse: it still puts DNS Client in GP mode, but contributes no +// usable rule, so our local catch-all is hidden until that empty parent is gone. // // Strategy (matching Tailscale's approach): // - Always write to the local path (baseline for non-domain machines). @@ -395,18 +399,20 @@ func otherGPRulesExist() bool { return false } -// cleanGPPath removes our CtrldCatchAll rule from the GP path and deletes -// the GP DnsPolicyConfig parent key if no other rules remain. Removing the -// empty GP key is critical: its mere existence forces DNS Client into "GP mode" -// where local-path rules are ignored. -func cleanGPPath() { +// cleanGPPath removes only ctrld's GP-path rule and deletes the GP parent when +// no rules remain. The return value tells callers whether the parent key was +// actually deleted, which means DNS Client should be signaled once. +// +// Do not leave an empty GP parent behind: Windows treats the parent key itself +// as the policy store boundary, so an empty key can still hide local-path rules. +func cleanGPPath() bool { // Delete our specific rule. registry.DeleteKey(registry.LOCAL_MACHINE, nrptBaseKey+`\`+nrptRuleName) // If the GP parent key is now empty, delete it entirely to exit "GP mode". k, err := registry.OpenKey(registry.LOCAL_MACHINE, nrptBaseKey, registry.ENUMERATE_SUB_KEYS) if err != nil { - return // Key doesn't exist — clean state. + return false // Key doesn't exist — clean state. } names, err := k.ReadSubKeyNames(-1) k.Close() @@ -414,12 +420,14 @@ func cleanGPPath() { if len(names) > 0 { mainLog.Load().Debug().Strs("remaining", names).Msg("DNS intercept: GP path has other rules, leaving parent key") } - return + return false } // Empty — delete it to exit "GP mode". if err := registry.DeleteKey(registry.LOCAL_MACHINE, nrptBaseKey); err == nil { mainLog.Load().Info().Msg("DNS intercept: deleted empty GP DnsPolicyConfig key (exits GP mode)") + return true } + return false } // writeNRPTRule writes a single NRPT catch-all rule at the given registry keyPath. @@ -542,10 +550,11 @@ func refreshNRPTPolicy() { // Group Policy refresh so NRPT changes take effect immediately. // Uses DnsFlushResolverCache from dnsapi.dll + RefreshPolicyEx from userenv.dll. func flushDNSCache() { - // Step 1: Refresh GP so DNS Client loads the new NRPT rules from registry. refreshNRPTPolicy() + flushDNSCacheOnly() +} - // Step 2: Flush the DNS cache so stale entries from pre-NRPT resolution are cleared. +func flushDNSCacheOnly() { if err := dnsapiDLL.Load(); err == nil { if err := procDnsFlushResolverCache.Find(); err == nil { ret, _, _ := procDnsFlushResolverCache.Call() @@ -555,7 +564,6 @@ func flushDNSCache() { } } } - // Fallback: use ipconfig /flushdns. if out, err := exec.Command("ipconfig", "/flushdns").CombinedOutput(); err != nil { mainLog.Load().Debug().Msgf("DNS intercept: ipconfig /flushdns failed: %v: %s", err, string(out)) } else { @@ -563,6 +571,12 @@ func flushDNSCache() { } } +func signalNRPTChange() { + refreshNRPTPolicy() + sendParamChange() + flushDNSCacheOnly() +} + // startDNSIntercept activates WFP-based DNS interception on Windows. // It creates a WFP sublayer and adds filters that block all outbound DNS (port 53) // traffic except to localhost (127.0.0.1/::1), ensuring all DNS queries must go @@ -598,21 +612,19 @@ func (p *prog) startDNSIntercept() error { mainLog.Load().Info().Msgf("DNS intercept: initializing (mode: %s)", interceptMode) logNRPTParentKeyState("pre-write") - // Two-phase empty parent key recovery: if the GP DnsPolicyConfig key exists - // but is empty, DNS Client has cached a "no rules" state and won't accept - // new rules even after they're written. Delete the empty key and signal DNS - // Client to reset before writing our rule. - // Two-phase recovery handles its own 2s signaling burst internally. - cleanEmptyNRPTParent() + // Empty parent key recovery: if the GP DnsPolicyConfig key exists but is + // empty, DNS Client enters GP mode and hides local rules. Delete empty + // parents first, then send one change signal so DNS Client drops stale state. + if cleanEmptyNRPTParent() { + signalNRPTChange() + } if err := addNRPTCatchAllRule(listenerIP); err != nil { return fmt.Errorf("dns intercept: failed to add NRPT catch-all rule: %w", err) } logNRPTParentKeyState("post-write") state.nrptActive = true - refreshNRPTPolicy() - sendParamChange() - flushDNSCache() + signalNRPTChange() mainLog.Load().Info().Msgf("DNS intercept: NRPT catch-all rule active — all DNS queries directed to %s", listenerIP) // Step 2: In hard mode, also set up WFP filters to block non-local DNS. @@ -1555,7 +1567,7 @@ func (p *prog) scheduleDelayedRechecks() { mainLog.Load().Error().Err(err).Msg("DNS intercept: failed to re-add NRPT catch-all rule") state.nrptActive = false } else { - flushDNSCache() + signalNRPTChange() mainLog.Load().Info().Msg("DNS intercept: NRPT catch-all rule restored") } } @@ -1593,14 +1605,22 @@ func (p *prog) nrptHealthMonitor(state *wfpState) { // Step 1: Check registry key exists. if !nrptCatchAllRuleExists() { + now := time.Now() + if ok, wait := state.nrptRecoveryLimiter.allow(now, p.cfg); !ok { + if state.nrptRecoveryLimiter.shouldLogSkip(now) { + mainLog.Load().Warn().Dur("remaining", wait). + Msg("DNS intercept: NRPT rule restore suppressed after repeated recovery flows") + } + continue + } mainLog.Load().Warn().Msg("DNS intercept: NRPT health check — catch-all rule missing, restoring") if err := addNRPTCatchAllRule(state.listenerIP); err != nil { mainLog.Load().Error().Err(err).Msg("DNS intercept: failed to restore NRPT catch-all rule") state.nrptActive = false continue } - refreshNRPTPolicy() - flushDNSCache() + signalNRPTChange() + state.nrptRecoveryLimiter.recordRecoveryFlow(time.Now(), p.cfg) mainLog.Load().Info().Msg("DNS intercept: NRPT catch-all rule restored by health monitor") // After restoring, verify it's actually working. go p.nrptProbeAndHeal() @@ -1612,6 +1632,8 @@ func (p *prog) nrptHealthMonitor(state *wfpState) { if !p.probeNRPT() { mainLog.Load().Warn().Msg("DNS intercept: NRPT health check — rule present but probe failed, running heal cycle") go p.nrptProbeAndHeal() + } else { + state.nrptRecoveryLimiter.recordStableSuccess() } // Step 3: In hard mode, also verify WFP sublayer. @@ -1710,43 +1732,39 @@ func sendParamChange() { } // cleanEmptyNRPTParent removes empty NRPT parent keys that block activation. -// An empty DnsPolicyConfig key (exists but no subkeys) causes DNS Client to -// cache "no rules" and ignore subsequently-added rules. +// Empty GP and local parents have different failure shapes: +// - empty GP parent: DNS Client is in GP mode and ignores local-path rules; +// - empty local parent: DNS Client can cache an empty local policy store. // -// Also cleans the GP path entirely if it has no non-ctrld rules, since the GP -// path's existence forces DNS Client into "GP mode" where it ignores the local -// service store path. +// This helper only changes registry state. The caller sends the single +// RefreshPolicyEx/paramchange/flush signal after it knows cleanup occurred. // -// Returns true if cleanup was performed (caller should add a delay). +// Returns true if cleanup was performed (caller should signal DNS Client). func cleanEmptyNRPTParent() bool { - cleaned := false - // Always clean the GP path — its existence blocks local path activation. - cleanGPPath() + cleaned := cleanGPPath() // Clean empty local/direct path parent key. - k, err := registry.OpenKey(registry.LOCAL_MACHINE, nrptDirectKey, registry.ENUMERATE_SUB_KEYS) - if err != nil { - return false - } - names, err := k.ReadSubKeyNames(-1) - k.Close() - if err != nil || len(names) > 0 { - return false + if !nrptParentKeyEmpty(nrptDirectKey) { + return cleaned } mainLog.Load().Warn().Msg("DNS intercept: found empty NRPT local parent key (blocks activation) — removing") if err := registry.DeleteKey(registry.LOCAL_MACHINE, nrptDirectKey); err != nil { mainLog.Load().Warn().Err(err).Msg("DNS intercept: failed to delete empty NRPT local parent key") - return false + return cleaned } - cleaned = true + return true +} - // Signal DNS Client to process the deletion and reset its internal cache. - mainLog.Load().Info().Msg("DNS intercept: empty NRPT parent key removed — signaling DNS Client") - sendParamChange() - flushDNSCache() - return cleaned +func nrptParentKeyEmpty(keyPath string) bool { + k, err := registry.OpenKey(registry.LOCAL_MACHINE, keyPath, registry.ENUMERATE_SUB_KEYS) + if err != nil { + return false + } + names, err := k.ReadSubKeyNames(-1) + k.Close() + return err == nil && len(names) == 0 } // logNRPTParentKeyState logs the state of both NRPT registry paths for diagnostics. @@ -1783,18 +1801,39 @@ func logNRPTParentKeyState(context string) { // nrptProbeAndHeal runs the NRPT probe with retries and escalating remediation. // Called asynchronously after startup and from the health monitor. // -// Retry sequence (each attempt: GP refresh + paramchange + flush → sleep → probe): -// 1. Immediate probe -// 2. GP refresh + paramchange + flush → 1s → probe -// 3. GP refresh + paramchange + flush → 2s → probe -// 4. GP refresh + paramchange + flush → 4s → probe +// Retry sequence: +// 1. Immediate probe. +// 2. If the GP parent is empty, clean it immediately, signal once, then probe. +// This is intentionally before the normal retry loop: policy refresh and +// Dnscache paramchange cannot make local rules visible while GP mode is +// selected by an empty GP parent. +// 3. Otherwise, signal DNS Client with increasing backoff between probes. func (p *prog) nrptProbeAndHeal() { + state, _ := p.dnsInterceptState.(*wfpState) + if state != nil { + now := time.Now() + if ok, wait := state.nrptRecoveryLimiter.allow(now, p.cfg); !ok { + if state.nrptRecoveryLimiter.shouldLogSkip(now) { + mainLog.Load().Warn().Dur("remaining", wait). + Msg("DNS intercept: NRPT recovery suppressed after repeated failed recovery flows") + } + return + } + } + if !nrptProbeRunning.CompareAndSwap(false, true) { mainLog.Load().Debug().Msg("DNS intercept: NRPT probe already running, skipping") return } defer nrptProbeRunning.Store(false) + remediated := false + defer func() { + if remediated && state != nil { + state.nrptRecoveryLimiter.recordRecoveryFlow(time.Now(), p.cfg) + } + }() + mainLog.Load().Info().Msg("DNS intercept: starting NRPT verification probe sequence") // Log parent key state for diagnostics. @@ -1805,17 +1844,37 @@ func (p *prog) nrptProbeAndHeal() { mainLog.Load().Info().Msg("DNS intercept: NRPT verified working") return } + remediated = true + + // If the GP parent exists but is empty, do not burn retries on Windows + // signaling. Those retries create SIEM noise but cannot succeed because DNS + // Client is still reading the empty GP store instead of the populated local + // store. Delete the blocker, send one notification, then re-probe. + if nrptParentKeyEmpty(nrptBaseKey) { + mainLog.Load().Warn().Msg("DNS intercept: NRPT probe failed with empty GP parent — cleaning before retry signaling") + if cleanEmptyNRPTParent() { + signalNRPTChange() + time.Sleep(1 * time.Second) + logNRPTParentKeyState("empty-gp-after-clean") + if p.probeNRPT() { + mainLog.Load().Info().Msg("DNS intercept: NRPT verified working after empty GP parent cleanup") + return + } + } + if nrptParentKeyEmpty(nrptBaseKey) { + mainLog.Load().Warn().Msg("DNS intercept: empty GP NRPT parent still present after cleanup; skipping redundant policy refresh retries") + return + } + } - // Attempts 2-4: GP refresh + paramchange + flush with increasing backoff + // Attempts 2-4: signal DNS Client with increasing backoff between probes. delays := []time.Duration{1 * time.Second, 2 * time.Second, 4 * time.Second} for i, delay := range delays { attempt := i + 2 mainLog.Load().Info().Int("attempt", attempt).Dur("delay", delay). - Msg("DNS intercept: NRPT probe failed, retrying with GP refresh + paramchange") + Msg("DNS intercept: NRPT probe failed, retrying with policy refresh + paramchange") logNRPTParentKeyState(fmt.Sprintf("probe-attempt-%d", attempt)) - refreshNRPTPolicy() - sendParamChange() - flushDNSCache() + signalNRPTChange() time.Sleep(delay) if p.probeNRPT() { mainLog.Load().Info().Int("attempt", attempt). @@ -1829,7 +1888,7 @@ func (p *prog) nrptProbeAndHeal() { // signal DNS Client to forget it, wait, then re-add and signal again. mainLog.Load().Warn().Msg("DNS intercept: all probes failed — attempting two-phase NRPT recovery (delete → signal → re-add)") listenerIP := "127.0.0.1" - if state, ok := p.dnsInterceptState.(*wfpState); ok { + if state != nil { listenerIP = state.listenerIP } @@ -1837,9 +1896,7 @@ func (p *prog) nrptProbeAndHeal() { _ = removeNRPTCatchAllRule() // If parent key is now empty after removing our rule, delete it too. cleanEmptyNRPTParent() - refreshNRPTPolicy() - sendParamChange() - flushDNSCache() + signalNRPTChange() logNRPTParentKeyState("nuclear-after-delete") // Wait for DNS Client to process the deletion. @@ -1850,9 +1907,7 @@ func (p *prog) nrptProbeAndHeal() { mainLog.Load().Error().Err(err).Msg("DNS intercept: failed to re-add NRPT after nuclear recovery") return } - refreshNRPTPolicy() - sendParamChange() - flushDNSCache() + signalNRPTChange() logNRPTParentKeyState("nuclear-after-readd") // Final probe after recovery. diff --git a/cmd/cli/dns_proxy.go b/cmd/cli/dns_proxy.go index fe37f609..4c08083a 100644 --- a/cmd/cli/dns_proxy.go +++ b/cmd/cli/dns_proxy.go @@ -1733,6 +1733,15 @@ func (p *prog) checkUpstreamOnce(upstream string, uc *ctrld.UpstreamConfig) erro mainLog.Load().Debug().Err(err).Msgf("Upstream %s check failed after %v (WFP loopback protect active)", upstream, duration) return errOsHealthcheckSuppressed } + // A no-route/network-unreachable failure means the endpoint's address + // family is available locally but unroutable (e.g. an IPv6 DoH endpoint + // while IPv6 is up but has no route). These repeat until the route + // returns and are handled by bounded backoff in the recovery loop, so + // keep them at debug to avoid sustained error-log spam. + if ctrldnet.IsUnreachable(err) { + mainLog.Load().Debug().Err(err).Msgf("Upstream %s check failed after %v (network unreachable)", upstream, duration) + return err + } mainLog.Load().Error().Err(err).Msgf("Upstream %s check failed after %v", upstream, duration) return err } @@ -1937,6 +1946,9 @@ func (p *prog) handleRecovery(reason RecoveryReason) { // 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 @@ -1948,9 +1960,10 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string defer wg.Done() mainLog.Load().Debug().Msgf("Starting recovery check loop for upstream: %s", name) attempts := 0 + unreachableStreak := 0 for { select { - case <-ctx.Done(): + case <-recoveryCtx.Done(): mainLog.Load().Debug().Msgf("Context canceled for upstream %s", name) return default: @@ -1962,13 +1975,30 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string select { case recoveredCh <- name: mainLog.Load().Debug().Msgf("Sent recovery notification for upstream %s", name) + cancel() default: mainLog.Load().Debug().Msg("Recovery channel full, another upstream already recovered") } return } - mainLog.Load().Debug().Msgf("Upstream %s check failed, sleeping before retry", name) - time.Sleep(checkUpstreamBackoffSleep) + // Back off the retry cadence for an unroutable endpoint so a + // host with IPv6 up but no route to the IPv6 DoH endpoint does + // not re-bootstrap/re-check every checkUpstreamBackoffSleep and + // spam the log. The backoff is bounded (checkUpstreamUnreachableBackoffMax) + // so the endpoint is still re-probed and recovers when the route + // returns; any other failure resets to the base cadence. + sleep := checkUpstreamBackoffSleep + if ctrldnet.IsUnreachable(err) { + unreachableStreak++ + sleep = unreachableRecoveryBackoff(unreachableStreak) + mainLog.Load().Debug().Msgf("Upstream %s unreachable (streak %d), backing off %s before retry", name, unreachableStreak, sleep) + } else { + unreachableStreak = 0 + mainLog.Load().Debug().Msgf("Upstream %s check failed, sleeping before retry", name) + } + if !sleepWithContext(recoveryCtx, sleep) { + 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 @@ -1996,6 +2026,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). diff --git a/cmd/cli/nrpt_recovery_limiter_windows.go b/cmd/cli/nrpt_recovery_limiter_windows.go new file mode 100644 index 00000000..64b6526a --- /dev/null +++ b/cmd/cli/nrpt_recovery_limiter_windows.go @@ -0,0 +1,101 @@ +//go:build windows + +package cli + +import ( + "sync" + "time" + + "github.com/Control-D-Inc/ctrld" +) + +const ( + // Default to current behavior: keep recovering indefinitely unless configured. + defaultNRPTRecoveryMaxAttempts = 0 + defaultNRPTRecoveryCooldown = 30 * time.Minute + + // Require more than one good health tick before clearing the circuit. A probe can + // pass briefly after delete/re-add even when another agent recreates broken NRPT state. + nrptRecoveryStableSuccessesToReset = 2 +) + +type nrptRecoveryLimiter struct { + mu sync.Mutex + attempts int + stableSuccesses int + cooldownUntil time.Time + lastSkipLog time.Time +} + +func nrptRecoveryMaxAttempts(cfg *ctrld.Config) int { + if cfg != nil && cfg.Service.NRPTRecoveryMaxAttempts != nil { + return *cfg.Service.NRPTRecoveryMaxAttempts + } + return defaultNRPTRecoveryMaxAttempts +} + +func nrptRecoveryCooldown(cfg *ctrld.Config) time.Duration { + if cfg != nil && cfg.Service.NRPTRecoveryCooldown != nil { + return *cfg.Service.NRPTRecoveryCooldown + } + return defaultNRPTRecoveryCooldown +} + +func (l *nrptRecoveryLimiter) allow(now time.Time, cfg *ctrld.Config) (bool, time.Duration) { + maxAttempts := nrptRecoveryMaxAttempts(cfg) + if maxAttempts <= 0 { + return true, 0 + } + + l.mu.Lock() + defer l.mu.Unlock() + + if now.Before(l.cooldownUntil) { + return false, l.cooldownUntil.Sub(now) + } + return true, 0 +} + +func (l *nrptRecoveryLimiter) recordRecoveryFlow(now time.Time, cfg *ctrld.Config) { + maxAttempts := nrptRecoveryMaxAttempts(cfg) + if maxAttempts <= 0 { + return + } + + cooldown := nrptRecoveryCooldown(cfg) + if cooldown <= 0 { + cooldown = defaultNRPTRecoveryCooldown + } + + l.mu.Lock() + defer l.mu.Unlock() + + l.stableSuccesses = 0 + l.attempts++ + if l.attempts >= maxAttempts { + l.cooldownUntil = now.Add(cooldown) + } +} + +func (l *nrptRecoveryLimiter) recordStableSuccess() { + l.mu.Lock() + defer l.mu.Unlock() + + l.stableSuccesses++ + if l.stableSuccesses >= nrptRecoveryStableSuccessesToReset { + l.attempts = 0 + l.cooldownUntil = time.Time{} + l.lastSkipLog = time.Time{} + } +} + +func (l *nrptRecoveryLimiter) shouldLogSkip(now time.Time) bool { + l.mu.Lock() + defer l.mu.Unlock() + + if l.lastSkipLog.IsZero() || now.Sub(l.lastSkipLog) >= 5*time.Minute { + l.lastSkipLog = now + return true + } + return false +} diff --git a/cmd/cli/nrpt_recovery_limiter_windows_test.go b/cmd/cli/nrpt_recovery_limiter_windows_test.go new file mode 100644 index 00000000..d0974a54 --- /dev/null +++ b/cmd/cli/nrpt_recovery_limiter_windows_test.go @@ -0,0 +1,74 @@ +//go:build windows + +package cli + +import ( + "testing" + "time" + + "github.com/Control-D-Inc/ctrld" +) + +func TestNRPTRecoveryLimiterCooldownAndStableReset(t *testing.T) { + maxAttempts := 2 + cooldown := 10 * time.Minute + cfg := &ctrld.Config{} + cfg.Service.NRPTRecoveryMaxAttempts = &maxAttempts + cfg.Service.NRPTRecoveryCooldown = &cooldown + + limiter := &nrptRecoveryLimiter{} + now := time.Unix(100, 0) + + if ok, wait := limiter.allow(now, cfg); !ok || wait != 0 { + t.Fatalf("initial allow = %v, %v; want true, 0", ok, wait) + } + + limiter.recordRecoveryFlow(now, cfg) + if ok, wait := limiter.allow(now.Add(time.Second), cfg); !ok || wait != 0 { + t.Fatalf("allow after first flow = %v, %v; want true, 0", ok, wait) + } + + limiter.recordRecoveryFlow(now.Add(2*time.Second), cfg) + if ok, wait := limiter.allow(now.Add(3*time.Second), cfg); ok || wait <= 0 { + t.Fatalf("allow after max flows = %v, %v; want false, positive wait", ok, wait) + } + + limiter.recordStableSuccess() + if ok, _ := limiter.allow(now.Add(4*time.Second), cfg); ok { + t.Fatal("one stable success cleared cooldown; want cooldown to remain") + } + + limiter.recordStableSuccess() + if ok, wait := limiter.allow(now.Add(5*time.Second), cfg); !ok || wait != 0 { + t.Fatalf("allow after stable reset = %v, %v; want true, 0", ok, wait) + } +} + +func TestNRPTRecoveryLimiterDefaultIsUnlimited(t *testing.T) { + cfg := &ctrld.Config{} + limiter := &nrptRecoveryLimiter{} + now := time.Unix(100, 0) + + for i := 0; i < 10; i++ { + limiter.recordRecoveryFlow(now.Add(time.Duration(i)*time.Second), cfg) + } + if ok, wait := limiter.allow(now.Add(time.Hour), cfg); !ok || wait != 0 { + t.Fatalf("default allow after recovery flows = %v, %v; want true, 0", ok, wait) + } +} + +func TestNRPTRecoveryLimiterUnlimited(t *testing.T) { + maxAttempts := 0 + cfg := &ctrld.Config{} + cfg.Service.NRPTRecoveryMaxAttempts = &maxAttempts + + limiter := &nrptRecoveryLimiter{} + now := time.Unix(100, 0) + + for i := 0; i < 10; i++ { + limiter.recordRecoveryFlow(now.Add(time.Duration(i)*time.Second), cfg) + } + if ok, wait := limiter.allow(now.Add(time.Hour), cfg); !ok || wait != 0 { + t.Fatalf("unlimited allow = %v, %v; want true, 0", ok, wait) + } +} diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index 5c5b2759..5da24906 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -34,6 +34,7 @@ import ( "github.com/Control-D-Inc/ctrld/internal/clientinfo" "github.com/Control-D-Inc/ctrld/internal/controld" "github.com/Control-D-Inc/ctrld/internal/dnscache" + ctrldnet "github.com/Control-D-Inc/ctrld/internal/net" "github.com/Control-D-Inc/ctrld/internal/router" "github.com/Control-D-Inc/ctrld/internal/router/dnsmasq" ) @@ -188,6 +189,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. // When non-empty, the DNS handler checks incoming queries against this value // and signals pfProbeCh if matched. The probe verifies that pf's rdr rules @@ -1328,13 +1344,14 @@ func errAddrInUse(err error) bool { var _ = errAddrInUse +// The unreachable winsock errnos (ENETUNREACH/EHOSTUNREACH) are matched via +// ctrldnet.IsUnreachable, which owns their definitions. +// // https://learn.microsoft.com/en-us/windows/win32/winsock/windows-sockets-error-codes-2 var ( windowsECONNREFUSED = syscall.Errno(10061) - windowsENETUNREACH = syscall.Errno(10051) windowsEINVAL = syscall.Errno(10022) windowsEADDRINUSE = syscall.Errno(10048) - windowsEHOSTUNREACH = syscall.Errno(10065) ) func errUrlNetworkError(err error) bool { @@ -1351,14 +1368,14 @@ func errNetworkError(err error) bool { if opErr.Temporary() { return true } + if ctrldnet.IsUnreachable(err) { + return true + } switch { case errors.Is(opErr.Err, syscall.ECONNREFUSED), errors.Is(opErr.Err, syscall.EINVAL), - errors.Is(opErr.Err, syscall.ENETUNREACH), - errors.Is(opErr.Err, windowsENETUNREACH), errors.Is(opErr.Err, windowsEINVAL), - errors.Is(opErr.Err, windowsECONNREFUSED), - errors.Is(opErr.Err, windowsEHOSTUNREACH): + errors.Is(opErr.Err, windowsECONNREFUSED): return true } } diff --git a/cmd/cli/prog_test.go b/cmd/cli/prog_test.go index 6831f5ae..562a8e03 100644 --- a/cmd/cli/prog_test.go +++ b/cmd/cli/prog_test.go @@ -1,7 +1,11 @@ package cli import ( + "context" + "net" + "net/url" "runtime" + "syscall" "testing" "time" @@ -12,6 +16,32 @@ 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 TestUnreachableRecoveryBackoff(t *testing.T) { + // Streak starts at the base cadence and doubles each attempt, capped at the max. + assert.Equal(t, checkUpstreamBackoffSleep, unreachableRecoveryBackoff(0)) + assert.Equal(t, checkUpstreamBackoffSleep, unreachableRecoveryBackoff(1)) + assert.Equal(t, 2*checkUpstreamBackoffSleep, unreachableRecoveryBackoff(2)) + assert.Equal(t, 4*checkUpstreamBackoffSleep, unreachableRecoveryBackoff(3)) + assert.Equal(t, checkUpstreamUnreachableBackoffMax, unreachableRecoveryBackoff(100)) +} + func Test_prog_dnsWatchdogEnabled(t *testing.T) { p := &prog{cfg: &ctrld.Config{}} diff --git a/cmd/cli/upstream_monitor.go b/cmd/cli/upstream_monitor.go index 6e19e38a..3eeb24b5 100644 --- a/cmd/cli/upstream_monitor.go +++ b/cmd/cli/upstream_monitor.go @@ -12,8 +12,27 @@ const ( maxFailureRequest = 50 // checkUpstreamBackoffSleep is the time interval between each upstream checks. checkUpstreamBackoffSleep = 2 * time.Second + // checkUpstreamUnreachableBackoffMax caps the recovery retry interval for an + // endpoint that keeps failing with a network-unreachable error. It bounds + // the backoff so an unroutable endpoint is still re-probed periodically and + // recovers once the route returns. + checkUpstreamUnreachableBackoffMax = 60 * time.Second ) +// unreachableRecoveryBackoff returns the retry interval for the given streak of +// consecutive network-unreachable failures. It starts at checkUpstreamBackoffSleep +// and doubles each attempt, capped at checkUpstreamUnreachableBackoffMax. +func unreachableRecoveryBackoff(streak int) time.Duration { + d := checkUpstreamBackoffSleep + for i := 1; i < streak; i++ { + d *= 2 + if d >= checkUpstreamUnreachableBackoffMax { + return checkUpstreamUnreachableBackoffMax + } + } + return d +} + // upstreamMonitor performs monitoring upstreams health. type upstreamMonitor struct { cfg *ctrld.Config diff --git a/cmd/cli/vpn_dns.go b/cmd/cli/vpn_dns.go index f6f8c20c..115ed80a 100644 --- a/cmd/cli/vpn_dns.go +++ b/cmd/cli/vpn_dns.go @@ -6,6 +6,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "github.com/rs/zerolog" "tailscale.com/net/netmon" @@ -50,6 +51,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 } @@ -69,6 +73,11 @@ func newVPNDNSManager(exemptFunc vpnDNSExemptFunc) *vpnDNSManager { // Called on network change events. func (m *vpnDNSManager) Refresh(guardAgainstNoNameservers bool) { logger := mainLog.Load() + if !m.refreshRunning.CompareAndSwap(false, true) { + logger.Debug().Msg("VPN DNS refresh already running, skipping duplicate") + return + } + defer m.refreshRunning.Store(false) logger.Debug().Msg("Refreshing VPN DNS configurations") discoverVPNDNS := m.discoverVPNDNS diff --git a/cmd/cli/vpn_dns_test.go b/cmd/cli/vpn_dns_test.go index 6a6d520d..846dc8ef 100644 --- a/cmd/cli/vpn_dns_test.go +++ b/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 diff --git a/config.go b/config.go index 1fc82e2f..fd94675e 100644 --- a/config.go +++ b/config.go @@ -241,6 +241,8 @@ type ServiceConfig struct { ForceRefetchWaitTime *int `mapstructure:"force_refetch_wait_time" toml:"force_refetch_wait_time,omitempty"` LeakOnUpstreamFailure *bool `mapstructure:"leak_on_upstream_failure" toml:"leak_on_upstream_failure,omitempty"` InterceptMode string `mapstructure:"intercept_mode" toml:"intercept_mode,omitempty" validate:"omitempty,oneof=off dns hard"` + NRPTRecoveryMaxAttempts *int `mapstructure:"nrpt_recovery_max_attempts" toml:"nrpt_recovery_max_attempts,omitempty" validate:"omitempty,gte=0"` + NRPTRecoveryCooldown *time.Duration `mapstructure:"nrpt_recovery_cooldown" toml:"nrpt_recovery_cooldown,omitempty"` Daemon bool `mapstructure:"-" toml:"-"` AllocateIP bool `mapstructure:"-" toml:"-"` } diff --git a/docs/config.md b/docs/config.md index 99e98c9c..8f21a4c2 100644 --- a/docs/config.md +++ b/docs/config.md @@ -295,6 +295,22 @@ If a remote upstream fails to resolve a query or is unreachable, `ctrld` will fo - Required: no - Default: true on Windows, MacOS and non-router Linux. +### nrpt_recovery_max_attempts +Windows DNS intercept mode uses NRPT health probes and recovery when Windows stops routing queries to the local `ctrld` listener. This limits how many consecutive recovery flows can run before `ctrld` enters a cooldown and stops making policy/Dnscache changes. + +Set to `0` to disable this circuit breaker and keep retrying indefinitely. + +- Type: integer +- Required: no +- Default: 0 (unlimited, current behavior) + +### nrpt_recovery_cooldown +Cooldown duration after `nrpt_recovery_max_attempts` consecutive Windows NRPT recovery flows. During cooldown, `ctrld` logs the suppressed recovery and avoids additional `RefreshPolicyEx`, Dnscache `paramchange`, and DNS cache flush calls. + +- Type: time duration string +- Required: no +- Default: 30m + ## Upstream The `[upstream]` section specifies the DNS upstream servers that `ctrld` will forward DNS requests to. diff --git a/docs/wfp-dns-intercept.md b/docs/wfp-dns-intercept.md index 2e7ece27..527d20d6 100644 --- a/docs/wfp-dns-intercept.md +++ b/docs/wfp-dns-intercept.md @@ -91,6 +91,20 @@ ctrld uses an adaptive strategy (matching [Tailscale's approach](https://github. the empty GP parent key. This ensures DNS Client stays in "local mode" where the local-path rule activates immediately via `paramchange`. +### Reproducing the Empty GP Parent Case + +This is a production code reference, so the temporary repro script is not kept in +the repository. For MR !942 review, the test script and exact before/after steps +are posted in the MR discussion. The scenario to compare is: + +1. Run the same approved PowerShell repro script against a pre-fix build and this + branch with the same ctrld config. +2. Create an empty GP NRPT parent key while ctrld is running in DNS intercept mode. +3. Confirm pre-fix logs can spend policy refresh/paramchange retries while the GP + parent remains empty. +4. Confirm post-fix logs clean the empty GP parent, send one NRPT-change signal, + and re-probe before normal retries. + ### VPN Coexistence NRPT uses most-specific-match. VPN NRPT rules for specific domains (e.g., diff --git a/internal/net/net.go b/internal/net/net.go index f4b55860..677e90dc 100644 --- a/internal/net/net.go +++ b/internal/net/net.go @@ -157,6 +157,101 @@ type parallelDialerResult struct { err error } +const ( + // unreachableBackoffBase is the initial suppression window applied to a + // dial address after it returns a network-unreachable error (e.g. + // "connect: no route to host"). The window grows exponentially up to + // unreachableBackoffMax on repeated failures, and is cleared as soon as + // the address dials successfully. + unreachableBackoffBase = 5 * time.Second + // unreachableBackoffMax caps the suppression window so an address is + // always re-probed within a bounded interval, preserving recovery when + // the route comes back. + unreachableBackoffMax = 60 * time.Second +) + +// Windows winsock codes for the unreachable errnos. A failing connect on +// Windows surfaces these raw WSA codes (WSAENETUNREACH/WSAEHOSTUNREACH), whereas +// syscall.ENETUNREACH/EHOSTUNREACH are Go's portable "invented" values +// (APPLICATION_ERROR + iota) that never equal them. Matching these explicitly is +// therefore required for the classifier to detect unreachable errors on Windows; +// errors.Is against the syscall.* constants alone would not. +// +// https://learn.microsoft.com/en-us/windows/win32/winsock/windows-sockets-error-codes-2 +var ( + windowsENETUNREACH = syscall.Errno(10051) + windowsEHOSTUNREACH = syscall.Errno(10065) +) + +// IsUnreachable reports whether err indicates the destination network or host +// has no route (ENETUNREACH/EHOSTUNREACH). These are the errors produced when +// an endpoint's address family is available locally but unroutable, e.g. an +// IPv6 DoH endpoint while the host has IPv6 but no route to it. +func IsUnreachable(err error) bool { + if err == nil { + return false + } + var opErr *net.OpError + if errors.As(err, &opErr) { + return errors.Is(opErr.Err, syscall.ENETUNREACH) || + errors.Is(opErr.Err, syscall.EHOSTUNREACH) || + errors.Is(opErr.Err, windowsENETUNREACH) || + errors.Is(opErr.Err, windowsEHOSTUNREACH) + } + return false +} + +type unreachableEntry struct { + until time.Time + backoff time.Duration +} + +// unreachableTracker records dial addresses that recently failed with a +// network-unreachable error so ParallelDialer can temporarily stop hammering +// them. This prevents an unroutable endpoint from generating a sustained dial +// /health-check storm, while still re-probing each address once its bounded +// backoff window expires so genuine recovery is never permanently blocked. +type unreachableTracker struct { + mu sync.Mutex + entries map[string]unreachableEntry +} + +var unreachable = &unreachableTracker{entries: make(map[string]unreachableEntry)} + +// suppressed reports whether addr is currently within its unreachable backoff +// window and should be skipped. +func (t *unreachableTracker) suppressed(addr string, now time.Time) bool { + t.mu.Lock() + defer t.mu.Unlock() + e, ok := t.entries[addr] + return ok && now.Before(e.until) +} + +// markUnreachable extends the suppression window for addr using a bounded +// exponential backoff. +func (t *unreachableTracker) markUnreachable(addr string, now time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + e := t.entries[addr] + if e.backoff == 0 { + e.backoff = unreachableBackoffBase + } else { + e.backoff *= 2 + if e.backoff > unreachableBackoffMax { + e.backoff = unreachableBackoffMax + } + } + e.until = now.Add(e.backoff) + t.entries[addr] = e +} + +// markReachable clears any suppression for addr after a successful dial. +func (t *unreachableTracker) markReachable(addr string) { + t.mu.Lock() + defer t.mu.Unlock() + delete(t.entries, addr) +} + type ParallelDialer struct { net.Dialer } @@ -165,26 +260,63 @@ func (d *ParallelDialer) DialContext(ctx context.Context, network string, addrs if len(addrs) == 0 { return nil, errors.New("empty addresses") } + + // Skip addresses that recently returned a network-unreachable error so an + // unroutable endpoint (e.g. an IPv6 DoH address while the host has IPv6 + // but no route to it) does not generate a sustained dial storm. Suppression + // is bounded: once an address's backoff window expires it is re-probed, so + // genuine recovery is preserved. + now := time.Now() + live := make([]string, 0, len(addrs)) + var suppressed int + for _, addr := range addrs { + if unreachable.suppressed(addr, now) { + suppressed++ + continue + } + live = append(live, addr) + } + if len(live) == 0 { + // Every candidate is within its unreachable backoff window. Fail fast + // and quietly instead of re-dialing known-unroutable addresses; the + // windows expire and re-probe shortly, so recovery still happens. + logger.Debug().Msgf("skipping %d unreachable address(es), all in backoff", suppressed) + // TODO: the errno here is hardcoded to EHOSTUNREACH, but the actual + // failure that triggered suppression may have been ENETUNREACH. This is + // harmless today (IsUnreachable treats both the same and nothing else + // inspects the errno), but if these errors are ever recorded/reported we + // should retain the real error in unreachableEntry and surface it here. + return nil, &net.OpError{Op: "dial", Net: network, Err: syscall.EHOSTUNREACH} + } + if suppressed > 0 { + logger.Debug().Msgf("skipping %d unreachable address(es) in backoff", suppressed) + } + ctx, cancel := context.WithCancel(ctx) defer cancel() done := make(chan struct{}) defer close(done) - ch := make(chan *parallelDialerResult, len(addrs)) + ch := make(chan *parallelDialerResult, len(live)) var wg sync.WaitGroup - wg.Add(len(addrs)) + wg.Add(len(live)) go func() { wg.Wait() close(ch) }() - for _, addr := range addrs { + for _, addr := range live { go func(addr string) { defer wg.Done() logger.Debug().Msgf("dialing to %s", addr) conn, err := d.Dialer.DialContext(ctx, network, addr) if err != nil { logger.Debug().Msgf("failed to dial %s: %v", addr, err) + if IsUnreachable(err) { + unreachable.markUnreachable(addr, time.Now()) + } + } else { + unreachable.markReachable(addr) } select { case ch <- ¶llelDialerResult{conn: conn, err: err}: diff --git a/internal/net/net_test.go b/internal/net/net_test.go index 7df3e09c..38187c5c 100644 --- a/internal/net/net_test.go +++ b/internal/net/net_test.go @@ -2,10 +2,77 @@ package net import ( "context" + "net" + "syscall" "testing" "time" ) +func TestIsUnreachable(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"enetunreach", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ENETUNREACH}, true}, + {"ehostunreach", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.EHOSTUNREACH}, true}, + {"windows enetunreach", &net.OpError{Op: "dial", Net: "tcp", Err: windowsENETUNREACH}, true}, + {"windows ehostunreach", &net.OpError{Op: "dial", Net: "tcp", Err: windowsEHOSTUNREACH}, true}, + {"connection refused", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ECONNREFUSED}, false}, + {"not an opError", syscall.ENETUNREACH, false}, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + if got := IsUnreachable(tc.err); got != tc.want { + t.Errorf("IsUnreachable(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +func TestUnreachableTracker(t *testing.T) { + tr := &unreachableTracker{entries: make(map[string]unreachableEntry)} + const addr = "[2606:1a40::22]:443" + now := time.Unix(0, 0) + + // Not suppressed before any failure. + if tr.suppressed(addr, now) { + t.Fatal("addr suppressed before any failure") + } + + // First failure suppresses for the base window. + tr.markUnreachable(addr, now) + if !tr.suppressed(addr, now.Add(unreachableBackoffBase-time.Millisecond)) { + t.Fatal("addr not suppressed within base backoff window") + } + if tr.suppressed(addr, now.Add(unreachableBackoffBase)) { + t.Fatal("addr still suppressed at end of base backoff window") + } + + // Backoff grows exponentially and is capped at the max. + tr.markUnreachable(addr, now) + if got := tr.entries[addr].backoff; got != 2*unreachableBackoffBase { + t.Fatalf("backoff after second failure = %v, want %v", got, 2*unreachableBackoffBase) + } + for i := 0; i < 10; i++ { + tr.markUnreachable(addr, now) + } + if got := tr.entries[addr].backoff; got != unreachableBackoffMax { + t.Fatalf("backoff not capped: got %v, want %v", got, unreachableBackoffMax) + } + + // A successful dial clears suppression entirely. + tr.markReachable(addr) + if tr.suppressed(addr, now) { + t.Fatal("addr still suppressed after markReachable") + } + if _, ok := tr.entries[addr]; ok { + t.Fatal("entry not removed after markReachable") + } +} + func TestProbeStackTimeout(t *testing.T) { done := make(chan struct{}) started := make(chan struct{}) diff --git a/resolver.go b/resolver.go index fbd2ad61..2284d04a 100644 --- a/resolver.go +++ b/resolver.go @@ -45,8 +45,12 @@ const ( const controldPublicDns = "76.76.2.0" +const maxConcurrentOSResolverExchanges = 128 + var controldPublicDnsWithPort = net.JoinHostPort(controldPublicDns, "53") +var osResolverExchangeSem = make(chan struct{}, maxConcurrentOSResolverExchanges) + var localResolver Resolver func init() { @@ -136,14 +140,15 @@ func availableNameservers() []string { // It's the caller's responsibility to ensure the system DNS is in a clean state before // calling this function. func InitializeOsResolver(guardAgainstNoNameservers bool) []string { + resolverMutex.Lock() + defer resolverMutex.Unlock() + nameservers := availableNameservers() // if no nameservers, return empty slice so we dont remove all nameservers if len(nameservers) == 0 && guardAgainstNoNameservers { return []string{} } ns := initializeOsResolver(nameservers) - resolverMutex.Lock() - defer resolverMutex.Unlock() or = newResolverWithNameserver(ns) return ns } @@ -466,6 +471,13 @@ func (o *osResolver) resolve(ctx context.Context, msg *dns.Msg) (*dns.Msg, error for _, server := range servers { go func(server string) { defer wg.Done() + release, ok := acquireOSResolverExchangeSlot(ctx) + if !ok { + ch <- &osResolverResult{err: ctx.Err(), server: server, lan: isLan} + return + } + defer release() + var answer *dns.Msg var err error var localOSResolverIP net.IP @@ -576,6 +588,15 @@ func (o *osResolver) resolve(ctx context.Context, msg *dns.Msg) (*dns.Msg, error return nil, errors.Join(errs...) } +func acquireOSResolverExchangeSlot(ctx context.Context) (func(), bool) { + select { + case osResolverExchangeSem <- struct{}{}: + return func() { <-osResolverExchangeSem }, true + case <-ctx.Done(): + return nil, false + } +} + func (o *osResolver) removeCache(key string) { o.cache.Delete(key) }