mirror of
https://github.com/Control-D-Inc/ctrld.git
synced 2026-10-12 01:58:46 +02:00
fix(darwin): probe interception after the host wakes
Rule text survives a suspend unchanged, so the watchdog reports the pf anchor intact while nothing translates, and the probe monitor is only armed by interface changes with a schedule frozen through the sleep. Detect the resume from the wall-clock gap and probe from there, with a bounded repair: at most two forced reloads across the retry schedule. Refresh the OS resolver before every probe - the probe aims at the first OS nameserver and passes when it has none, so a pre-sleep list makes it meaningless. Stand down for stabilization, which verifies on completion.
This commit is contained in:
5 files changed
+972
-28
No files matched your search
+227
-28
@@ -209,6 +209,44 @@ func setCtrldGroupID(gid int) error {
|
||||
func (p *prog) startDNSIntercept() error {
|
||||
mainLog.Load().Info().Msg("DNS intercept: initializing macOS packet filter (pf) redirect")
|
||||
|
||||
if err := installPFInterceptFn(p); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p.dnsInterceptState = &pfState{
|
||||
anchorFile: pfAnchorFile,
|
||||
anchorName: pfAnchorName,
|
||||
}
|
||||
|
||||
// Store the initial set of tunnel interfaces so we can detect changes later.
|
||||
p.mu.Lock()
|
||||
p.lastTunnelIfaces = discoverTunnelInterfacesForReconcile()
|
||||
p.pendingTunnelIfaces = nil
|
||||
p.hasPendingTunnelIfaces = false
|
||||
p.mu.Unlock()
|
||||
|
||||
lc := p.cfg.FirstListener()
|
||||
if lc != nil {
|
||||
mainLog.Load().Info().Msgf("DNS intercept: pf redirect active - all outbound DNS (port 53) redirected to %s:%d via anchor %q", lc.IP, lc.Port, pfAnchorName)
|
||||
} else {
|
||||
mainLog.Load().Info().Msgf("DNS intercept: pf redirect active - all outbound DNS (port 53) redirected via anchor %q", pfAnchorName)
|
||||
}
|
||||
|
||||
p.startInterceptBackgroundWork()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// installPFInterceptFn is the privileged-install seam. Everything behind it needs root,
|
||||
// a live pf and a writable /etc/pf.anchors - which is what kept startDNSIntercept, and
|
||||
// with it the handoff to the background watchers, out of reach of every test.
|
||||
var installPFInterceptFn = (*prog).installPFIntercept
|
||||
|
||||
// installPFIntercept performs the privileged part of starting interception: validation,
|
||||
// _ctrld group setup, anchor file write, pfctl load, enable, and post-load verification.
|
||||
// It returns nil only once pf is authoritatively active, so the caller may publish
|
||||
// intercept mode; on failure it has already rolled back whatever it applied.
|
||||
func (p *prog) installPFIntercept() error {
|
||||
if err := p.validateDNSIntercept(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -316,32 +354,33 @@ func (p *prog) startDNSIntercept() error {
|
||||
return fmt.Errorf("dns intercept: post-load PF verification failed")
|
||||
}
|
||||
|
||||
p.dnsInterceptState = &pfState{
|
||||
anchorFile: pfAnchorFile,
|
||||
anchorName: pfAnchorName,
|
||||
}
|
||||
|
||||
// Store the initial set of tunnel interfaces so we can detect changes later.
|
||||
p.mu.Lock()
|
||||
p.lastTunnelIfaces = discoverTunnelInterfacesForReconcile()
|
||||
p.pendingTunnelIfaces = nil
|
||||
p.hasPendingTunnelIfaces = false
|
||||
p.mu.Unlock()
|
||||
|
||||
lc := p.cfg.FirstListener()
|
||||
if lc != nil {
|
||||
mainLog.Load().Info().Msgf("DNS intercept: pf redirect active — all outbound DNS (port 53) redirected to %s:%d via anchor %q", lc.IP, lc.Port, pfAnchorName)
|
||||
} else {
|
||||
mainLog.Load().Info().Msgf("DNS intercept: pf redirect active — all outbound DNS (port 53) redirected via anchor %q", pfAnchorName)
|
||||
}
|
||||
|
||||
// Start the pf watchdog to detect and restore rules if another program
|
||||
// (e.g., Windscribe desktop, macOS configd) replaces the pf ruleset.
|
||||
go p.pfWatchdog()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// pfWatchdogFn and runSuspendWatcherFn are the background-work seams. Tests use them to
|
||||
// observe what startInterceptBackgroundWork registers without running either loop, which
|
||||
// would otherwise reach pfctl.
|
||||
var (
|
||||
pfWatchdogFn = (*prog).pfWatchdog
|
||||
runSuspendWatcherFn = runSuspendWatcher
|
||||
)
|
||||
|
||||
// startInterceptBackgroundWork launches the goroutines that guard an active intercept.
|
||||
//
|
||||
// Split out of startDNSIntercept so the registration is reachable from a test:
|
||||
// startDNSIntercept needs root, a live pf and a writable /etc/pf.anchors, so nothing
|
||||
// covers this wiring if it lives inline there.
|
||||
func (p *prog) startInterceptBackgroundWork() {
|
||||
// Detect and restore rules if another program (e.g., Windscribe desktop, macOS
|
||||
// configd) replaces the pf ruleset.
|
||||
go pfWatchdogFn(p)
|
||||
|
||||
// A host sleep freezes every timer above it, and rule text survives the suspend
|
||||
// unchanged, so no existing path notices that pf stopped translating while the host
|
||||
// was asleep. Watch for the resume itself and probe from there.
|
||||
go runSuspendWatcherFn(p.stopCh, p.verifyInterceptAfterWake)
|
||||
}
|
||||
|
||||
func (p *prog) rollbackDNSInterceptStart() {
|
||||
if out, err := exec.Command("pfctl", "-a", pfAnchorName, "-F", "all").CombinedOutput(); err != nil {
|
||||
mainLog.Load().Warn().Err(err).Msgf("DNS intercept: startup rollback could not flush anchor (output: %s)", strings.TrimSpace(string(out)))
|
||||
@@ -518,6 +557,10 @@ func (p *prog) stopDNSIntercept() error {
|
||||
}
|
||||
}
|
||||
p.pfDelayedRecheckTimers = nil
|
||||
if p.pfSettleFollowupTimer != nil {
|
||||
p.pfSettleFollowupTimer.Stop()
|
||||
p.pfSettleFollowupTimer = nil
|
||||
}
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
|
||||
for !p.pfEnsureRunning.CompareAndSwap(false, true) {
|
||||
@@ -1525,6 +1568,146 @@ func (p *prog) verifyInterceptAfterStabilization() {
|
||||
mainLog.Load().Error().Msg("DNS intercept: interception still not translating after the post-stabilization reload — the watchdog will retry")
|
||||
}
|
||||
|
||||
// pfWakeProbeDelays is the bounded retry schedule for the post-resume functional probe.
|
||||
// Each value is the wait before its own attempt, not an offset from the resume, so the
|
||||
// attempts land at +0s, +2s, +6s and +14s after the suspend is noticed. A resumed host
|
||||
// brings its link, default route, resolver list and pf translation state back over
|
||||
// several seconds, so a single probe at +0 decides too early. A var so tests can shorten
|
||||
// it.
|
||||
var pfWakeProbeDelays = []time.Duration{0, 2 * time.Second, 4 * time.Second, 8 * time.Second}
|
||||
|
||||
// pfWakeMaxRepairs bounds forced reloads across the whole schedule. QA on the post-wake
|
||||
// outage saw one forced reload fail to restore translation and a later one succeed, so a
|
||||
// single attempt is not enough - but this stays a bounded repair rather than a reload
|
||||
// loop, because every reload flushes pf state and kills in-flight DoH connections.
|
||||
const pfWakeMaxRepairs = 2
|
||||
|
||||
// pfWakeProbeOwnerWait bounds how long one attempt waits for another prober to release
|
||||
// functional-probe ownership. Short on purpose: a holder that is genuinely probing is
|
||||
// doing this path's job, and there are further attempts left in the schedule.
|
||||
var pfWakeProbeOwnerWait = 250 * time.Millisecond
|
||||
|
||||
// verifyInterceptAfterWake proves pf is actually translating after the host resumes from
|
||||
// sleep, and repairs it a bounded number of times when it is not.
|
||||
//
|
||||
// Nothing else covers this window. The periodic watchdog compares rule text, which
|
||||
// survives a suspend unchanged, so it reports the anchor intact while every query through
|
||||
// the system resolver times out. The interception probe monitor is only armed by
|
||||
// interface changes and its backoff schedule is frozen through the suspend, so whether it
|
||||
// probes after a resume depends on where its timers happened to be when the host slept.
|
||||
// QA measured 18s of no public DNS on a host whose link and default route were back
|
||||
// within the first second, recovered only because an unrelated VPN reconnect happened to
|
||||
// drive the post-stabilization verifier; without that reconnect nothing would have probed.
|
||||
//
|
||||
// The OS resolver list is refreshed before every probe. probePFIntercept aims at the
|
||||
// first OS nameserver and reports success when it has no target, so a list still holding
|
||||
// a pre-sleep VPN resolver turns this into a meaningless pass or a false failure.
|
||||
func (p *prog) verifyInterceptAfterWake(gap time.Duration) {
|
||||
if p.dnsInterceptState == nil {
|
||||
return
|
||||
}
|
||||
mainLog.Load().Info().Msgf("DNS intercept: host resumed after a %s gap - verifying interception is still translating", gap.Round(time.Second))
|
||||
|
||||
repairs := 0
|
||||
probed := 0
|
||||
for attempt, delay := range pfWakeProbeDelays {
|
||||
if !p.waitBeforeWakeProbe(delay) {
|
||||
return
|
||||
}
|
||||
if p.dnsInterceptState == nil {
|
||||
return
|
||||
}
|
||||
// Stabilization owns pf while a VPN's ruleset settles, and runs its own
|
||||
// functional verification when it finishes. Probing or reloading underneath it
|
||||
// is the mutual overwriting stabilization exists to prevent.
|
||||
if p.pfStabilizing.Load() {
|
||||
mainLog.Load().Debug().Msg("DNS intercept: post-wake probe standing down - stabilization owns pf and verifies on completion")
|
||||
return
|
||||
}
|
||||
if p.pfExecBackoffActive() {
|
||||
mainLog.Load().Debug().Msg("DNS intercept: post-wake probe deferred this attempt - pf exec backoff active")
|
||||
continue
|
||||
}
|
||||
// Eligibility first, ownership second, and release before the next delay: a
|
||||
// prober that holds the flag while sleeping starves the post-stabilization
|
||||
// verifier, which reads a held flag as "somebody is probing".
|
||||
if !p.claimFunctionalProbeOwner(pfWakeProbeOwnerWait) {
|
||||
mainLog.Load().Debug().Msg("DNS intercept: post-wake probe skipped this attempt - another prober holds ownership")
|
||||
continue
|
||||
}
|
||||
restored, repaired := p.runWakeProbeAttempt(attempt, repairs)
|
||||
probed++
|
||||
p.pfMonitorRunning.Store(false)
|
||||
if repaired {
|
||||
repairs++
|
||||
}
|
||||
if restored {
|
||||
return
|
||||
}
|
||||
}
|
||||
// Report what actually ran: an attempt deferred by pf exec backoff or skipped for
|
||||
// want of probe ownership never probes, so the schedule length would claim
|
||||
// verification that did not happen - and with every attempt deferred, nothing was
|
||||
// measured at all.
|
||||
if probed == 0 {
|
||||
mainLog.Load().Warn().Msg("DNS intercept: post-wake verification never probed - every attempt was deferred or skipped - the watchdog will retry")
|
||||
return
|
||||
}
|
||||
mainLog.Load().Error().Msgf("DNS intercept: interception still not translating after resume (%d probe attempts, %d forced reloads) - the watchdog will retry",
|
||||
probed, repairs)
|
||||
}
|
||||
|
||||
// waitBeforeWakeProbe waits out one attempt's delay, reporting false when the service
|
||||
// stopped while waiting.
|
||||
func (p *prog) waitBeforeWakeProbe(delay time.Duration) bool {
|
||||
if delay <= 0 {
|
||||
return true
|
||||
}
|
||||
timer := time.NewTimer(delay)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-p.stopCh:
|
||||
return false
|
||||
case <-timer.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// runWakeProbeAttempt runs one probe of the post-resume schedule, forcing a reload when
|
||||
// the probe fails and repairs so far leave budget for one. It reports whether
|
||||
// interception is translating and whether this attempt spent a repair. A reload that
|
||||
// could not run does not count: nothing was changed, so the next attempt may try again.
|
||||
// The caller holds functional-probe ownership for the call.
|
||||
func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired bool) {
|
||||
initializeOsResolver(true)
|
||||
if probePFInterceptFn(p) {
|
||||
if attempt > 0 || repairs > 0 {
|
||||
mainLog.Load().Info().Msgf("DNS intercept: post-wake probe %d/%d passed - interception is translating", attempt+1, len(pfWakeProbeDelays))
|
||||
} else {
|
||||
mainLog.Load().Debug().Msg("DNS intercept: post-wake probe passed - interception is translating")
|
||||
}
|
||||
return true, false
|
||||
}
|
||||
|
||||
if repairs >= pfWakeMaxRepairs {
|
||||
mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - repair budget of %d forced reloads is spent, not reloading again",
|
||||
attempt+1, len(pfWakeProbeDelays), pfWakeMaxRepairs)
|
||||
return false, false
|
||||
}
|
||||
|
||||
mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - rules survived the suspend but pf is not translating, forcing a reload",
|
||||
attempt+1, len(pfWakeProbeDelays))
|
||||
if !forceReloadPFInterceptFn(p) {
|
||||
mainLog.Load().Error().Msg("DNS intercept: post-wake forced reload did not run")
|
||||
return false, false
|
||||
}
|
||||
if probePFInterceptFn(p) {
|
||||
mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-wake reload")
|
||||
return true, true
|
||||
}
|
||||
return false, true
|
||||
}
|
||||
|
||||
var runPFAnchorCheckCommand = func(args ...string) ([]byte, error) {
|
||||
return exec.Command("pfctl", args...).CombinedOutput()
|
||||
}
|
||||
@@ -1687,18 +1870,34 @@ func (p *prog) dnsInterceptIgnoredChangeReconcileDue(now time.Time) bool {
|
||||
}
|
||||
}
|
||||
|
||||
// scheduleDNSAfterVPNSettleRefresh queues one follow-up VPN DNS refresh.
|
||||
//
|
||||
// The timer is tracked rather than fired and forgotten. Teardown has to be able to
|
||||
// cancel it, and VPN churn can finish stabilization several times in a row, where
|
||||
// untracked timers stack up into repeats of the same scutil/VPN-DNS work.
|
||||
func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Duration) {
|
||||
timer := time.AfterFunc(delay, func() {
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
defer p.pfDelayedRecheckMu.Unlock()
|
||||
if p.pfSettleFollowupTimer != nil {
|
||||
p.pfSettleFollowupTimer.Stop()
|
||||
}
|
||||
p.pfSettleFollowupTimer = time.AfterFunc(delay, func() {
|
||||
if p.dnsInterceptState == nil {
|
||||
return
|
||||
}
|
||||
p.refreshDNSAfterVPNSettle(reason)
|
||||
})
|
||||
// Track the timer like the other delayed rechecks, so intercept teardown
|
||||
// (and test cleanup) can stop it instead of letting it fire afterwards.
|
||||
}
|
||||
|
||||
// stopPFSettleFollowup cancels a pending post-settle refresh. A callback that already
|
||||
// fired still checks dnsInterceptState, so either teardown order is safe.
|
||||
func (p *prog) stopPFSettleFollowup() {
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
p.pfDelayedRecheckTimers = append(p.pfDelayedRecheckTimers, timer)
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
defer p.pfDelayedRecheckMu.Unlock()
|
||||
if p.pfSettleFollowupTimer != nil {
|
||||
p.pfSettleFollowupTimer.Stop()
|
||||
p.pfSettleFollowupTimer = nil
|
||||
}
|
||||
}
|
||||
|
||||
func (p *prog) pfExecBackoffActive() bool {
|
||||
|
||||
@@ -1000,6 +1000,16 @@ func TestFinishPFStabilizationRunsFunctionalVerification(t *testing.T) {
|
||||
t.Errorf("forced reloads = %d, want 1", *reloads)
|
||||
}
|
||||
|
||||
stopPFTestTimers(p)
|
||||
}
|
||||
|
||||
// stopPFTestTimers cancels every timer a stabilization pass can leave pending.
|
||||
//
|
||||
// The whole test binary shares one unsynchronised log sink, so a timer that outlives its
|
||||
// test logs concurrently with whichever test is running next and trips the race detector
|
||||
// there - reported against the innocent test. The post-settle follow-up is a 4s timer, so
|
||||
// the collision lands far from its origin.
|
||||
func stopPFTestTimers(p *prog) {
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
timers := append([]*time.Timer(nil), p.pfDelayedRecheckTimers...)
|
||||
p.pfDelayedRecheckTimers = nil
|
||||
@@ -1007,4 +1017,416 @@ func TestFinishPFStabilizationRunsFunctionalVerification(t *testing.T) {
|
||||
for _, timer := range timers {
|
||||
timer.Stop()
|
||||
}
|
||||
p.stopPFSettleFollowup()
|
||||
}
|
||||
|
||||
// TestScheduleDNSAfterVPNSettleRefreshIsCancellable pins the tracking itself. An
|
||||
// untracked follow-up cannot be cancelled by teardown or by this helper, which is how a
|
||||
// 4s timer escaped its test and raced the log sink of a later one.
|
||||
func TestScheduleDNSAfterVPNSettleRefreshIsCancellable(t *testing.T) {
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.scheduleDNSAfterVPNSettleRefresh("test", time.Hour)
|
||||
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
tracked := p.pfSettleFollowupTimer != nil
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
if !tracked {
|
||||
t.Fatal("the follow-up refresh was scheduled without being tracked; nothing can cancel it")
|
||||
}
|
||||
|
||||
p.stopPFSettleFollowup()
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
cleared := p.pfSettleFollowupTimer == nil
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
if !cleared {
|
||||
t.Error("stopPFSettleFollowup left the timer in place")
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduleDNSAfterVPNSettleRefreshReplacesPending keeps VPN churn from stacking
|
||||
// repeats of the same scutil/VPN-DNS refresh.
|
||||
func TestScheduleDNSAfterVPNSettleRefreshReplacesPending(t *testing.T) {
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.scheduleDNSAfterVPNSettleRefresh("first", time.Hour)
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
first := p.pfSettleFollowupTimer
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
|
||||
p.scheduleDNSAfterVPNSettleRefresh("second", time.Hour)
|
||||
p.pfDelayedRecheckMu.Lock()
|
||||
second := p.pfSettleFollowupTimer
|
||||
p.pfDelayedRecheckMu.Unlock()
|
||||
defer p.stopPFSettleFollowup()
|
||||
|
||||
if first == second {
|
||||
t.Fatal("the second schedule reused the first timer")
|
||||
}
|
||||
if first.Stop() {
|
||||
t.Error("the superseded timer was still armed; stabilization churn would stack refreshes")
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// post-wake functional verification tests
|
||||
// =============================================================================
|
||||
|
||||
// stubWakeProbeSchedule shortens the post-resume retry schedule and neutralises the OS
|
||||
// resolver refresh, so the bounded-repair contract can be tested without real delays or
|
||||
// a real scutil call. It returns the number of resolver refreshes performed.
|
||||
func stubWakeProbeSchedule(t *testing.T, attempts int) (refreshes *int) {
|
||||
t.Helper()
|
||||
originalDelays, originalWait := pfWakeProbeDelays, pfWakeProbeOwnerWait
|
||||
originalResolver := initializeOsResolver
|
||||
t.Cleanup(func() {
|
||||
pfWakeProbeDelays, pfWakeProbeOwnerWait = originalDelays, originalWait
|
||||
initializeOsResolver = originalResolver
|
||||
})
|
||||
|
||||
pfWakeProbeDelays = make([]time.Duration, attempts)
|
||||
for i := range pfWakeProbeDelays {
|
||||
pfWakeProbeDelays[i] = time.Millisecond
|
||||
}
|
||||
pfWakeProbeOwnerWait = 10 * time.Millisecond
|
||||
|
||||
calls := 0
|
||||
initializeOsResolver = func(bool) []string {
|
||||
calls++
|
||||
return []string{"10.0.0.1:53"}
|
||||
}
|
||||
return &calls
|
||||
}
|
||||
|
||||
// TestWakeProbeDetectsInterceptionBrokenBySuspend is the hole this closes. Rule text
|
||||
// survives a suspend unchanged, so the watchdog reports the anchor intact, and the probe
|
||||
// monitor is only armed by interface changes with a schedule frozen through the sleep.
|
||||
// Nothing probed after the resume, and QA measured 18s of no public DNS on a host whose
|
||||
// link and default route were already back.
|
||||
func TestWakeProbeDetectsInterceptionBrokenBySuspend(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 4)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes != 2 {
|
||||
t.Errorf("probe calls = %d, want 2: one to detect and one to confirm the repair", *probes)
|
||||
}
|
||||
if *reloads != 1 {
|
||||
t.Errorf("forced reloads = %d, want 1", *reloads)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbePassStopsImmediately keeps a healthy resume free of pf work: a reload
|
||||
// flushes pf state and kills in-flight DoH connections for nothing.
|
||||
func TestWakeProbePassStopsImmediately(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 4)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes != 1 {
|
||||
t.Errorf("probe calls = %d, want 1: a passing probe must end the schedule", *probes)
|
||||
}
|
||||
if *reloads != 0 {
|
||||
t.Errorf("forced reloads = %d, want 0 on a healthy resume", *reloads)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeRetriesAcrossTheSchedule covers the observed shape directly: the first
|
||||
// forced reload did not restore translation and a later one did, so a single attempt is
|
||||
// not enough. The second repair must still happen, and the confirming pass must end it.
|
||||
func TestWakeProbeRetriesAcrossTheSchedule(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 4)
|
||||
// attempt 1: probe fails, reload, confirm fails.
|
||||
// attempt 2: probe fails, reload, confirm passes.
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false, false, false, true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *reloads != 2 {
|
||||
t.Errorf("forced reloads = %d, want 2: one failed repair must not end the schedule", *reloads)
|
||||
}
|
||||
if *probes != 4 {
|
||||
t.Errorf("probe calls = %d, want 4", *probes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeRepairsAreBounded pins the bound. A probe that never passes must exhaust
|
||||
// the repair budget and then keep probing without reloading, so this can never become the
|
||||
// unbounded reload loop the issue rules out.
|
||||
func TestWakeProbeRepairsAreBounded(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 6)
|
||||
probes, reloads := stubStabilizationProbe(t, nil, true) // every probe fails
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *reloads != pfWakeMaxRepairs {
|
||||
t.Errorf("forced reloads = %d, want the budget of %d", *reloads, pfWakeMaxRepairs)
|
||||
}
|
||||
// Two repaired attempts probe twice each; the remaining four probe once each.
|
||||
if want := 2*pfWakeMaxRepairs + (6 - pfWakeMaxRepairs); *probes != want {
|
||||
t.Errorf("probe calls = %d, want %d", *probes, want)
|
||||
}
|
||||
if p.pfMonitorRunning.Load() {
|
||||
t.Error("functional-probe ownership was left held after the schedule finished")
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeRefreshesResolverBeforeEveryProbe is why a stale list matters: the probe
|
||||
// aims at the first OS nameserver and reports success when it has none, so a list still
|
||||
// holding a pre-sleep VPN resolver turns the probe into a meaningless pass or a false
|
||||
// failure. The captured run showed the resolver set changing after the resume.
|
||||
func TestWakeProbeRefreshesResolverBeforeEveryProbe(t *testing.T) {
|
||||
refreshes := stubWakeProbeSchedule(t, 3)
|
||||
stubStabilizationProbe(t, nil, false) // probes fail, reload refuses to run
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *refreshes != 3 {
|
||||
t.Errorf("OS resolver refreshes = %d, want one per attempt (3)", *refreshes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeStandsDownForStabilization keeps the two repair paths from fighting over
|
||||
// pf. Stabilization owns the ruleset while a VPN settles and runs its own functional
|
||||
// verification when it finishes; reloading underneath it is the mutual overwriting
|
||||
// stabilization exists to prevent.
|
||||
func TestWakeProbeStandsDownForStabilization(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 4)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.pfStabilizing.Store(true)
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes != 0 || *reloads != 0 {
|
||||
t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 while stabilization owns pf", *probes, *reloads)
|
||||
}
|
||||
if !p.claimFunctionalProbeOwner(0) {
|
||||
t.Error("standing down took functional-probe ownership; the post-stabilization verifier would skip")
|
||||
}
|
||||
p.pfMonitorRunning.Store(false)
|
||||
}
|
||||
|
||||
// TestWakeProbeYieldsToAnotherProberThenRetries is the ownership handoff. A probe monitor
|
||||
// that is genuinely working owns this attempt, but the schedule must not be spent waiting
|
||||
// on it: the next attempt still has to probe once ownership is free.
|
||||
func TestWakeProbeYieldsToAnotherProberThenRetries(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 3)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.pfMonitorRunning.Store(true)
|
||||
go func() {
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
p.pfMonitorRunning.Store(false)
|
||||
}()
|
||||
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes == 0 {
|
||||
t.Fatal("the schedule gave up entirely because another prober held ownership at first")
|
||||
}
|
||||
if *reloads != 1 {
|
||||
t.Errorf("forced reloads = %d, want 1", *reloads)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeSkippedDuringExecBackoff protects the resource-exhaustion guard: each
|
||||
// probe forks a helper, which is exactly what backoff exists to stop.
|
||||
func TestWakeProbeSkippedDuringExecBackoff(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 3)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true)
|
||||
|
||||
p := &prog{dnsInterceptState: &pfState{}}
|
||||
p.pfExecBackoffUntil.Store(time.Now().Add(time.Minute).UnixMilli())
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes != 0 || *reloads != 0 {
|
||||
t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 during pf exec backoff", *probes, *reloads)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeSkippedWhenInterceptDisabled keeps the watcher harmless when intercept
|
||||
// mode is off: the detector runs for the process, the repair is intercept-only.
|
||||
func TestWakeProbeSkippedWhenInterceptDisabled(t *testing.T) {
|
||||
stubWakeProbeSchedule(t, 3)
|
||||
probes, reloads := stubStabilizationProbe(t, []bool{false}, true)
|
||||
|
||||
p := &prog{}
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
|
||||
if *probes != 0 || *reloads != 0 {
|
||||
t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 with intercept disabled", *probes, *reloads)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeProbeStopsWhenServiceStops keeps a resume during shutdown from holding the
|
||||
// schedule open past the stop request.
|
||||
func TestWakeProbeStopsWhenServiceStops(t *testing.T) {
|
||||
originalDelays := pfWakeProbeDelays
|
||||
pfWakeProbeDelays = []time.Duration{0, time.Hour}
|
||||
t.Cleanup(func() { pfWakeProbeDelays = originalDelays })
|
||||
originalResolver := initializeOsResolver
|
||||
initializeOsResolver = func(bool) []string { return []string{"10.0.0.1:53"} }
|
||||
t.Cleanup(func() { initializeOsResolver = originalResolver })
|
||||
|
||||
probes, _ := stubStabilizationProbe(t, nil, false) // first probe fails, reload refuses
|
||||
|
||||
stopCh := make(chan struct{})
|
||||
close(stopCh)
|
||||
p := &prog{dnsInterceptState: &pfState{}, stopCh: stopCh}
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
p.verifyInterceptAfterWake(18 * time.Second)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("verifyInterceptAfterWake ignored the stop request and waited out its schedule")
|
||||
}
|
||||
if *probes != 1 {
|
||||
t.Errorf("probe calls = %d, want 1 before the stop was observed", *probes)
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// post-wake watcher registration tests
|
||||
// =============================================================================
|
||||
|
||||
// TestStartInterceptBackgroundWorkRegistersTheWakeVerifier covers the caller wiring. The
|
||||
// detector and the wake-repair path can both be correct while nothing connects them, and
|
||||
// then a host sleep is never noticed in production.
|
||||
func TestStartInterceptBackgroundWorkRegistersTheWakeVerifier(t *testing.T) {
|
||||
type registration struct {
|
||||
stopCh <-chan struct{}
|
||||
onResume func(time.Duration)
|
||||
}
|
||||
registered := make(chan registration, 1)
|
||||
watchdogStarted := make(chan struct{}, 1)
|
||||
originalWatcher, originalWatchdog := runSuspendWatcherFn, pfWatchdogFn
|
||||
runSuspendWatcherFn = func(stopCh <-chan struct{}, onResume func(time.Duration)) {
|
||||
registered <- registration{stopCh: stopCh, onResume: onResume}
|
||||
}
|
||||
// Both loops are stubbed rather than run: each would reach pfctl, and stopping the
|
||||
// real watchdog means closing stopCh, which the wake verifier reads as "service
|
||||
// stopping" and would abandon its probe schedule before the first probe.
|
||||
pfWatchdogFn = func(*prog) { watchdogStarted <- struct{}{} }
|
||||
t.Cleanup(func() { runSuspendWatcherFn, pfWatchdogFn = originalWatcher, originalWatchdog })
|
||||
|
||||
stubWakeProbeSchedule(t, 4)
|
||||
probes, _ := stubStabilizationProbe(t, []bool{false, true}, true)
|
||||
|
||||
// Open for the whole test: an active intercept is not stopping.
|
||||
stopCh := make(chan struct{})
|
||||
p := &prog{dnsInterceptState: &pfState{}, stopCh: stopCh}
|
||||
p.startInterceptBackgroundWork()
|
||||
|
||||
select {
|
||||
case <-watchdogStarted:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Error("no pf watchdog was started; a replaced pf ruleset would never be restored")
|
||||
}
|
||||
|
||||
var reg registration
|
||||
select {
|
||||
case reg = <-registered:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("no suspend watcher was started; a host sleep would go unnoticed")
|
||||
}
|
||||
if reg.stopCh != stopCh {
|
||||
t.Error("the watcher was not wired to the service stop channel; it would outlive the intercept")
|
||||
}
|
||||
|
||||
// Prove the callback is the post-wake verifier and not some other function: a
|
||||
// reported resume has to probe interception. How many probes and repairs that
|
||||
// schedule is worth belongs to TestWakeProbeDetectsInterceptionBrokenBySuspend;
|
||||
// pinning it again here would only couple this test to that tuning.
|
||||
reg.onResume(18 * time.Second)
|
||||
if *probes == 0 {
|
||||
t.Error("the resume callback probed nothing; it is not wired to verifyInterceptAfterWake")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStartDNSInterceptStartsBackgroundWork closes the caller-wiring gap. Everything the
|
||||
// privileged install does sits behind installPFInterceptFn, so this drives the real
|
||||
// startDNSIntercept and checks what a successful start hands off: without the handoff, an
|
||||
// active intercept runs with no pf watchdog and no post-wake verification, and every
|
||||
// detector, recovery and registration test still passes.
|
||||
func TestStartDNSInterceptStartsBackgroundWork(t *testing.T) {
|
||||
installed := 0
|
||||
originalInstall := installPFInterceptFn
|
||||
installPFInterceptFn = func(*prog) error { installed++; return nil }
|
||||
originalDiscover := discoverTunnelInterfacesForReconcile
|
||||
discoverTunnelInterfacesForReconcile = func() []string { return nil }
|
||||
|
||||
watchdogStarted := make(chan struct{}, 1)
|
||||
watcherStarted := make(chan struct{}, 1)
|
||||
originalWatchdog, originalWatcher := pfWatchdogFn, runSuspendWatcherFn
|
||||
pfWatchdogFn = func(*prog) { watchdogStarted <- struct{}{} }
|
||||
runSuspendWatcherFn = func(<-chan struct{}, func(time.Duration)) { watcherStarted <- struct{}{} }
|
||||
t.Cleanup(func() {
|
||||
installPFInterceptFn = originalInstall
|
||||
discoverTunnelInterfacesForReconcile = originalDiscover
|
||||
pfWatchdogFn, runSuspendWatcherFn = originalWatchdog, originalWatcher
|
||||
})
|
||||
|
||||
p := &prog{
|
||||
cfg: &ctrld.Config{Listener: map[string]*ctrld.ListenerConfig{"0": {IP: "127.0.0.1", Port: 53}}},
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
if err := p.startDNSIntercept(); err != nil {
|
||||
t.Fatalf("startDNSIntercept() = %v, want nil", err)
|
||||
}
|
||||
if installed != 1 {
|
||||
t.Errorf("privileged install ran %d times, want 1", installed)
|
||||
}
|
||||
if p.dnsInterceptState == nil {
|
||||
t.Fatal("a successful start did not publish intercept state")
|
||||
}
|
||||
|
||||
select {
|
||||
case <-watchdogStarted:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Error("startDNSIntercept did not start the pf watchdog; a replaced pf ruleset would never be restored")
|
||||
}
|
||||
select {
|
||||
case <-watcherStarted:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Error("startDNSIntercept did not start the suspend watcher; a host sleep would go unnoticed")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStartDNSInterceptSkipsBackgroundWorkWhenInstallFails keeps the handoff conditional:
|
||||
// watchers over an intercept that was never installed would repair state nothing owns.
|
||||
func TestStartDNSInterceptSkipsBackgroundWorkWhenInstallFails(t *testing.T) {
|
||||
originalInstall := installPFInterceptFn
|
||||
installPFInterceptFn = func(*prog) error { return errors.New("pf unavailable") }
|
||||
originalWatchdog, originalWatcher := pfWatchdogFn, runSuspendWatcherFn
|
||||
pfWatchdogFn = func(*prog) { t.Error("the pf watchdog started after a failed install") }
|
||||
runSuspendWatcherFn = func(<-chan struct{}, func(time.Duration)) {
|
||||
t.Error("the suspend watcher started after a failed install")
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
installPFInterceptFn = originalInstall
|
||||
pfWatchdogFn, runSuspendWatcherFn = originalWatchdog, originalWatcher
|
||||
})
|
||||
|
||||
p := &prog{stopCh: make(chan struct{})}
|
||||
if err := p.startDNSIntercept(); err == nil {
|
||||
t.Fatal("startDNSIntercept() = nil after the privileged install failed")
|
||||
}
|
||||
if p.dnsInterceptState != nil {
|
||||
t.Error("a failed start published intercept state; the caller would skip its DNS fallback")
|
||||
}
|
||||
// The goroutines above are started before this returns if at all, but give a failing
|
||||
// wiring a moment to report rather than racing the test's end.
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
@@ -250,6 +250,12 @@ type prog struct {
|
||||
pfDelayedRecheckMu sync.Mutex //lint:ignore U1000 used on darwin
|
||||
pfDelayedRecheckTimers []*time.Timer //lint:ignore U1000 used on darwin
|
||||
|
||||
// pfSettleFollowupTimer is the pending post-settle VPN DNS refresh, if any.
|
||||
// Tracked so teardown can cancel it and a later stabilization replaces it
|
||||
// instead of stacking another copy of the same refresh. Protected by
|
||||
// pfDelayedRecheckMu.
|
||||
pfSettleFollowupTimer *time.Timer //lint:ignore U1000 used on darwin
|
||||
|
||||
// pfIgnoredChangeLastReconcile bounds immediate pf/VPN-DNS work for noisy
|
||||
// ignored macOS network deltas. Tunnel changes bypass this limit, and the
|
||||
// existing delayed checks provide a trailing reconciliation after churn.
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
package cli
|
||||
|
||||
import "time"
|
||||
|
||||
// Suspend/resume detection without an OS notification API.
|
||||
//
|
||||
// A host sleep suspends the whole process: timers stop advancing and resume where they
|
||||
// left off, so a tick scheduled two seconds out can arrive a minute later. That
|
||||
// overshoot is the only in-process evidence that the host slept, and nothing else tells
|
||||
// the DNS-intercept repair paths that the state they proved before sleep - rule text,
|
||||
// resolver list, upstream connections - may no longer hold.
|
||||
|
||||
const (
|
||||
// suspendProbeInterval is how often the detector measures elapsed wall-clock time.
|
||||
// Short enough that a resume is noticed within a couple of seconds, which is what
|
||||
// the post-wake continuity contract measures from.
|
||||
suspendProbeInterval = 2 * time.Second
|
||||
|
||||
// suspendGapThreshold is the elapsed wall clock over one interval that counts as a
|
||||
// suspend. A pfctl storm or a loaded host can stretch a tick, so this sits at
|
||||
// several times the interval; a real sleep overshoots by far more. Erring low is
|
||||
// cheap on purpose: a false positive costs one probe, and a stall this long is
|
||||
// worth a probe either way.
|
||||
suspendGapThreshold = 8 * time.Second
|
||||
)
|
||||
|
||||
// wallNanoNow reads the wall clock in nanoseconds.
|
||||
//
|
||||
// The detector works in wall-clock nanoseconds rather than time.Time on purpose.
|
||||
// time.Time.Sub prefers the monotonic reading whenever both operands carry one, and
|
||||
// Darwin's monotonic clock stops while the host is asleep - so a monotonic difference
|
||||
// across a suspend is just the tick interval and hides the very gap this looks for.
|
||||
// Taking int64 wall nanos makes that mistake unrepresentable rather than a comment.
|
||||
func wallNanoNow() int64 { return time.Now().UnixNano() }
|
||||
|
||||
// suspendDetector turns observations of a fixed-interval tick into suspend/resume
|
||||
// events. It measures elapsed time rather than counting ticks, because a suspended
|
||||
// process gets one late tick, not the ticks it missed.
|
||||
type suspendDetector struct {
|
||||
threshold time.Duration
|
||||
wallNano func() int64
|
||||
lastNano int64
|
||||
}
|
||||
|
||||
// newSuspendDetector returns a detector that reports gaps of at least threshold.
|
||||
// wallNano is a seam for tests; nil means the real wall clock.
|
||||
func newSuspendDetector(threshold time.Duration, wallNano func() int64) *suspendDetector {
|
||||
if wallNano == nil {
|
||||
wallNano = wallNanoNow
|
||||
}
|
||||
return &suspendDetector{threshold: threshold, wallNano: wallNano, lastNano: wallNano()}
|
||||
}
|
||||
|
||||
// observe records a tick and reports the elapsed time when it reaches the threshold,
|
||||
// which means the process was not running for most of that span.
|
||||
func (d *suspendDetector) observe() (time.Duration, bool) {
|
||||
current := d.wallNano()
|
||||
gap := time.Duration(current - d.lastNano)
|
||||
d.lastNano = current
|
||||
if gap < d.threshold {
|
||||
return 0, false
|
||||
}
|
||||
return gap, true
|
||||
}
|
||||
|
||||
// reset re-baselines the detector to the current wall clock. Time that has already
|
||||
// passed is deliberately discarded rather than measured: use it after doing work
|
||||
// outside the sampling loop, which is not a suspend even though it looks like one.
|
||||
func (d *suspendDetector) reset() { d.lastNano = d.wallNano() }
|
||||
|
||||
// watchForResume calls onResume once per detected suspend/resume gap, until stopCh
|
||||
// closes or tick is closed. tick is a parameter so tests can drive it directly.
|
||||
//
|
||||
// onResume runs on this goroutine, so recovery is serialized against detection - one
|
||||
// resume cannot start a second recovery on top of the first. The cost is that no
|
||||
// sampling happens while it runs, which the re-baseline below accounts for.
|
||||
func watchForResume(stopCh <-chan struct{}, tick <-chan time.Time, d *suspendDetector, onResume func(gap time.Duration)) {
|
||||
for {
|
||||
select {
|
||||
case <-stopCh:
|
||||
return
|
||||
case _, ok := <-tick:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if gap, resumed := d.observe(); resumed {
|
||||
onResume(gap)
|
||||
// Re-baseline: the recovery just ran, and its own runtime exceeds the
|
||||
// threshold on its own (the post-wake probe schedule waits for seconds
|
||||
// before a single probe timeout or reload). The ticker keeps firing and
|
||||
// buffers a tick while it works, so without this the next observe()
|
||||
// measures the callback and reports it as a fresh suspend - which starts
|
||||
// another recovery, which is again long enough to look like a suspend.
|
||||
// Under a persistent pf failure that is an unbounded reload loop, and
|
||||
// each pass flushes pf state and kills in-flight DoH connections.
|
||||
d.reset()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runSuspendWatcher watches for suspend/resume against a real ticker.
|
||||
//
|
||||
//lint:ignore U1000 used on darwin (post-wake interception verification)
|
||||
func runSuspendWatcher(stopCh <-chan struct{}, onResume func(gap time.Duration)) {
|
||||
ticker := time.NewTicker(suspendProbeInterval)
|
||||
defer ticker.Stop()
|
||||
watchForResume(stopCh, ticker.C, newSuspendDetector(suspendGapThreshold, nil), onResume)
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fakeWallClock returns the queued offsets from a fixed base, then repeats the last one.
|
||||
type fakeWallClock struct {
|
||||
nanos []int64
|
||||
calls int
|
||||
}
|
||||
|
||||
func (c *fakeWallClock) now() int64 {
|
||||
if c.calls < len(c.nanos) {
|
||||
n := c.nanos[c.calls]
|
||||
c.calls++
|
||||
return n
|
||||
}
|
||||
return c.nanos[len(c.nanos)-1]
|
||||
}
|
||||
|
||||
// wallClockAt builds a clock whose readings are the given offsets from an arbitrary
|
||||
// epoch. The first reading is consumed by newSuspendDetector; the rest are ticks.
|
||||
func wallClockAt(offsets ...time.Duration) *fakeWallClock {
|
||||
const base = int64(1787081656_000000000) // 2026-08-18T14:34:16-05:00
|
||||
nanos := make([]int64, 0, len(offsets))
|
||||
for _, off := range offsets {
|
||||
nanos = append(nanos, base+int64(off))
|
||||
}
|
||||
return &fakeWallClock{nanos: nanos}
|
||||
}
|
||||
|
||||
func TestSuspendDetectorIgnoresOrdinaryTicks(t *testing.T) {
|
||||
// Construction, then three ticks two seconds apart, one stretched to 3s by a
|
||||
// loaded host. None of these is a suspend.
|
||||
clock := wallClockAt(0, 2*time.Second, 4*time.Second, 7*time.Second)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
if gap, resumed := d.observe(); resumed {
|
||||
t.Errorf("tick %d reported a suspend of %s; ordinary jitter must not count", i, gap)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuspendDetectorReportsTheGapOnce replays the captured incident: the last pre-sleep
|
||||
// sample at 14:34:47 and the first post-resume one at 14:35:05.
|
||||
func TestSuspendDetectorReportsTheGapOnce(t *testing.T) {
|
||||
clock := wallClockAt(
|
||||
31*time.Second, // construction, 14:34:47
|
||||
33*time.Second, // ordinary tick
|
||||
49*time.Second, // 14:35:05 - a 16s gap over a 2s interval
|
||||
51*time.Second, // ordinary again
|
||||
)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
if _, resumed := d.observe(); resumed {
|
||||
t.Fatal("the tick before the suspend reported a resume")
|
||||
}
|
||||
gap, resumed := d.observe()
|
||||
if !resumed {
|
||||
t.Fatal("a 16s gap over a 2s interval was not reported as a suspend")
|
||||
}
|
||||
if gap != 16*time.Second {
|
||||
t.Errorf("gap = %s, want 16s: the reported gap must be the measured wall clock", gap)
|
||||
}
|
||||
if _, resumed := d.observe(); resumed {
|
||||
t.Error("the resume was reported twice; the baseline must advance to the late tick")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuspendDetectorThresholdCoversShortNaps guards the lower bound. A nap only a few
|
||||
// seconds longer than the interval still breaks pf translation, so the threshold must not
|
||||
// drift up to a value that only catches long sleeps.
|
||||
func TestSuspendDetectorThresholdCoversShortNaps(t *testing.T) {
|
||||
if suspendGapThreshold > 10*time.Second {
|
||||
t.Errorf("suspendGapThreshold = %s; a short nap would go unnoticed", suspendGapThreshold)
|
||||
}
|
||||
if suspendGapThreshold <= 2*suspendProbeInterval {
|
||||
t.Errorf("suspendGapThreshold = %s is too close to the %s interval; ordinary jitter would fire it",
|
||||
suspendGapThreshold, suspendProbeInterval)
|
||||
}
|
||||
|
||||
clock := wallClockAt(0, suspendGapThreshold)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
if _, resumed := d.observe(); !resumed {
|
||||
t.Error("a gap exactly at the threshold was not reported")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSuspendDetectorIgnoresBackwardClockSteps(t *testing.T) {
|
||||
// An NTP step backwards must not read as a resume, and must not leave the baseline
|
||||
// in the future either: the next ordinary tick still has to look ordinary.
|
||||
clock := wallClockAt(0, -30*time.Second, -28*time.Second)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
if gap, resumed := d.observe(); resumed {
|
||||
t.Errorf("a backward clock step reported a suspend of %s", gap)
|
||||
}
|
||||
if gap, resumed := d.observe(); resumed {
|
||||
t.Errorf("the tick after a backward step reported a suspend of %s", gap)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatchForResumeCallsBackPerGap(t *testing.T) {
|
||||
clock := wallClockAt(
|
||||
0,
|
||||
2*time.Second, // ordinary
|
||||
40*time.Second, // resume: 38s
|
||||
42*time.Second, // ordinary
|
||||
99*time.Second, // second resume: 57s
|
||||
)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
tick := make(chan time.Time, 4)
|
||||
for i := 0; i < 4; i++ {
|
||||
tick <- time.Time{}
|
||||
}
|
||||
close(tick)
|
||||
|
||||
var gaps []time.Duration
|
||||
watchForResume(nil, tick, d, func(gap time.Duration) { gaps = append(gaps, gap) })
|
||||
|
||||
if len(gaps) != 2 {
|
||||
t.Fatalf("onResume called %d times (%v), want 2", len(gaps), gaps)
|
||||
}
|
||||
if gaps[0] != 38*time.Second || gaps[1] != 57*time.Second {
|
||||
t.Errorf("gaps = %v, want [38s 57s]", gaps)
|
||||
}
|
||||
}
|
||||
|
||||
// advancingClock is a wall clock the test moves by hand, so a callback's runtime can be
|
||||
// simulated without sleeping for it.
|
||||
type advancingClock struct{ nano int64 }
|
||||
|
||||
func (c *advancingClock) now() int64 { return c.nano }
|
||||
func (c *advancingClock) advance(d time.Duration) { c.nano += int64(d) }
|
||||
|
||||
// TestWatchForResumeDoesNotRetriggerOnItsOwnCallback pins the loop this closes. onResume
|
||||
// runs on the watcher goroutine, and the real callback - verifyInterceptAfterWake - takes
|
||||
// longer than the gap threshold by design: its configured waits alone total more than the
|
||||
// threshold, before any probe timeout or forced reload. The ticker buffers a tick while it
|
||||
// works, so a detector that keeps its pre-callback baseline measures the callback itself as
|
||||
// the next gap and starts another recovery, with another repair budget, indefinitely under
|
||||
// a persistent pf failure.
|
||||
func TestWatchForResumeDoesNotRetriggerOnItsOwnCallback(t *testing.T) {
|
||||
clock := &advancingClock{}
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
// Capacity 1, like time.Ticker's channel: exactly one tick can queue up behind a
|
||||
// callback in production.
|
||||
tick := make(chan time.Time, 1)
|
||||
calls := 0
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
watchForResume(nil, tick, d, func(time.Duration) {
|
||||
calls++
|
||||
if calls > 1 {
|
||||
// A retrigger - the failure this test reports. There is nothing left to
|
||||
// simulate, and tick is already closed.
|
||||
return
|
||||
}
|
||||
// The recovery takes longer than the threshold, and a tick lands while it
|
||||
// runs. Closing here ends the watcher after it drains that queued tick,
|
||||
// which is the observation under test.
|
||||
clock.advance(2 * suspendGapThreshold)
|
||||
tick <- time.Time{}
|
||||
close(tick)
|
||||
})
|
||||
}()
|
||||
|
||||
clock.advance(2 * suspendGapThreshold) // the genuine suspend
|
||||
tick <- time.Time{}
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("watchForResume did not finish; the queued tick was never consumed")
|
||||
}
|
||||
|
||||
if calls != 1 {
|
||||
t.Errorf("onResume called %d times, want 1: the callback's own runtime was measured as a second suspend", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatchForResumeStopsOnStopCh(t *testing.T) {
|
||||
stopCh := make(chan struct{})
|
||||
tick := make(chan time.Time) // never fires
|
||||
clock := wallClockAt(0, time.Hour)
|
||||
d := newSuspendDetector(suspendGapThreshold, clock.now)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
watchForResume(stopCh, tick, d, func(time.Duration) {
|
||||
t.Error("onResume ran without a tick")
|
||||
})
|
||||
close(done)
|
||||
}()
|
||||
|
||||
close(stopCh)
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("watchForResume did not return once stopCh was closed")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user