Compare commits

...

4 Commits

Author SHA1 Message Date
Cuong Manh Le d7f43ea4bf Merge pull request #323 from Control-D-Inc/release-branch-v1.5.4
Release v1.5.4
2026-07-14 21:06:20 +07:00
Dev Scribe 41ca69849a Add Windows NRPT recovery circuit breaker
Windows DNS intercept mode runs an NRPT health monitor that restores the
catch-all rule and re-signals DNS Client whenever Windows stops routing
queries to the local listener. When another agent (MDM, VPN, GPO) keeps
putting NRPT back into a broken state, that loop never converges: ctrld
repeatedly calls RefreshPolicyEx, Dnscache paramchange, and flushes the
DNS cache, producing continuous flash writes and SIEM noise while never
fixing anything.

Add a recovery limiter that trips after a configurable number of
consecutive recovery flows and enters a cooldown, during which recovery
is suppressed (logged at most once every 5 minutes). Clearing the
circuit requires two consecutive stable health successes rather than
one, because a probe can pass briefly right after delete/re-add even
when the underlying NRPT state is still broken.

New [service] options gate the behavior and default to the previous
unlimited behavior:
  - nrpt_recovery_max_attempts (default 0 = unlimited)
  - nrpt_recovery_cooldown     (default 30m)

Also collapse the repeated refresh + paramchange + flush sequence into a
single signalNRPTChange() helper, and make cleanGPPath /
cleanEmptyNRPTParent only mutate the registry and report whether cleanup
happened, so callers send exactly one DNS Client change signal instead
of several. When the GP DnsPolicyConfig parent exists but is empty,
nrptProbeAndHeal now cleans it and signals once before spending the
normal policy-refresh retry budget, since those retries cannot succeed
while DNS Client is stuck in GP mode.

Add unit tests for the limiter's cooldown, stable-reset, and unlimited
paths, and document the new options and the empty-GP repro.
2026-07-14 01:12:37 +07:00
Cuong Manh Le 0d8df38dc1 fix: back off unroutable IPv6 DoH upstream health-check spam
When IPv6 is available locally but the selected IPv6 DoH endpoint is
unroutable (e.g. dialing [2606:1a40::22]:443 returns "no route to host"
while IPv4 stays usable), ctrld re-bootstrapped and re-dialed the endpoint
every ~2s. A weekend soak produced ~46.7k "no route to host" lines, with
the dial/health-check loop dominating the log during bad windows.

Add bounded backoff/suppression for network-unreachable endpoints at two
levels:

- ParallelDialer (internal/net): track dial addresses that fail with
  ENETUNREACH/EHOSTUNREACH and skip them for an exponentially growing,
  bounded window (5s -> 60s). A successful dial clears the entry
  immediately, so recovery is preserved when the route returns. When every
  candidate is suppressed the dial fails fast and quietly instead of
  hammering known-unroutable addresses.

- Upstream recovery loop (cmd/cli): demote unreachable check failures to
  debug and back off the retry cadence (2s -> 60s) for an unreachable
  streak; any other failure resets to the base cadence.

The new IsUnreachable classifier lives in internal/net and is reused by
cmd/cli's errNetworkError, so the unreachable-errno matching has a single
definition. Note the explicit winsock constants (10051/10065) are required
on Windows: syscall.ENETUNREACH/EHOSTUNREACH are Go's portable "invented"
values and never equal the raw WSA codes a failing connect surfaces.

Suppression and backoff are always bounded, so IPv6 is never disabled until
restart and recovers on its own once the route is back. Split-stack
selection and the #549 macOS intercept recovery work are untouched.

Adds unit tests for the classifier, the dialer's suppression tracker, and
the recovery backoff schedule.
2026-07-14 01:12:34 +07:00
Dev Scribe 3ef17bc5b9 fix: back off macOS pf watchdog exec storms 2026-07-14 01:09:14 +07:00
17 changed files with 839 additions and 81 deletions
+84 -1
View File
@@ -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.
@@ -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)
}
}
+45
View File
@@ -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)
}
})
}
}
+121 -66
View File
@@ -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 cleaned
}
return true
}
func nrptParentKeyEmpty(keyPath string) bool {
k, err := registry.OpenKey(registry.LOCAL_MACHINE, keyPath, registry.ENUMERATE_SUB_KEYS)
if err != nil {
return false
}
cleaned = 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
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
// Attempts 2-4: GP refresh + paramchange + flush with increasing backoff
// 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: 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.
+44 -3
View File
@@ -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).
+101
View File
@@ -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
}
@@ -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)
}
}
+23 -6
View File
@@ -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
}
}
+30
View File
@@ -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{}}
+19
View File
@@ -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
+9
View File
@@ -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
+32
View File
@@ -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
+2
View File
@@ -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:"-"`
}
+16
View File
@@ -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.
+14
View File
@@ -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.,
+135 -3
View File
@@ -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 <- &parallelDialerResult{conn: conn, err: err}:
+67
View File
@@ -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{})
+23 -2
View File
@@ -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)
}