Merge pull request #323 from Control-D-Inc/release-branch-v1.5.4

Release v1.5.4
This commit is contained in:
Cuong Manh Le
2026-07-14 21:06:20 +07:00
committed by GitHub
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)
}