diff --git a/wgengine/watchdog.go b/wgengine/watchdog.go index 19505be896989..06624b301d622 100644 --- a/wgengine/watchdog.go +++ b/wgengine/watchdog.go @@ -6,17 +6,20 @@ package wgengine import ( + "errors" "fmt" "log" "net/netip" "runtime/pprof" "strings" "sync" + "sync/atomic" "time" "tailscale.com/envknob" "tailscale.com/ipn/ipnstate" "tailscale.com/net/dns" + "tailscale.com/net/packet" "tailscale.com/tailcfg" "tailscale.com/types/key" "tailscale.com/types/netmap" @@ -26,6 +29,10 @@ import ( "tailscale.com/wgengine/wgcfg" ) +// ErrWatchdogTimeout is returned when a watchdog configured with a timeout +// callback times out. +var ErrWatchdogTimeout = errors.New("wgengine watchdog timeout") + // NewWatchdog wraps an Engine and makes sure that all methods complete // within a reasonable amount of time. // @@ -34,12 +41,32 @@ func NewWatchdog(e Engine) Engine { if envknob.Bool("TS_DEBUG_DISABLE_WATCHDOG") { return e } + return newWatchdog(e, nil, false) +} + +// NewWatchdogWithTimeoutCallback wraps an Engine and synchronously calls +// callback when the first operation times out. The callback must not block or +// call methods on the returned Engine. +// Unlike NewWatchdog, this does not terminate the process on timeout. +func NewWatchdogWithTimeoutCallback(e Engine, callback func(operation string)) Engine { + if envknob.Bool("TS_DEBUG_DISABLE_WATCHDOG") { + return e + } + return newWatchdog(e, callback, true) +} + +func newWatchdog(e Engine, callback func(operation string), recoverOnTimeout bool) *watchdogEngine { return &watchdogEngine{ - wrap: e, - logf: log.Printf, - fatalf: log.Fatalf, - maxWait: 45 * time.Second, - inFlight: make(map[inFlightKey]time.Time), + wrap: e, + logf: log.Printf, + fatalf: log.Fatalf, + timeoutCallback: callback, + recoverOnTimeout: recoverOnTimeout, + maxWait: 45 * time.Second, + poisonedDone: make(chan struct{}), + closeDone: make(chan struct{}), + wrappedWaitDone: make(chan struct{}), + inFlight: make(map[inFlightKey]time.Time), } } @@ -48,12 +75,44 @@ type inFlightKey struct { ctr uint64 } +type whoIsIPPortResult struct { + tsIP netip.Addr + ok bool +} + +type peerForIPResult struct { + peer PeerForIP + ok bool +} + +func watchdogValue[T any](e *watchdogEngine, name string, fn func() T) (T, bool) { + var zero T + resultCh := make(chan T, 1) + err := e.watchdogErr(name, func() error { + resultCh <- fn() + return nil + }) + if err != nil { + return zero, false + } + return <-resultCh, true +} + type watchdogEngine struct { wrap Engine logf func(format string, args ...any) fatalf func(format string, args ...any) maxWait time.Duration + timeoutCallback func(operation string) + recoverOnTimeout bool + poisoned atomic.Bool + poisonedDone chan struct{} + closeStarted atomic.Bool + closeDone chan struct{} + waitStarted atomic.Bool + wrappedWaitDone chan struct{} + // Track the start time(s) of in-flight operations inFlightMu sync.Mutex inFlight map[inFlightKey]time.Time @@ -61,6 +120,19 @@ type watchdogEngine struct { } func (e *watchdogEngine) watchdogErr(name string, fn func() error) error { + err, _ := e.watchdogErrStarted(name, fn) + return err +} + +func (e *watchdogEngine) watchdogErrStarted(name string, fn func() error) (error, bool) { + if e.poisoned.Load() { + if e.recoverOnTimeout { + <-e.poisonedDone + return ErrWatchdogTimeout, false + } + return nil, false + } + // Track all in-flight operations so we can print more useful error // messages on watchdog failure e.inFlightMu.Lock() @@ -78,38 +150,71 @@ func (e *watchdogEngine) watchdogErr(name string, fn func() error) error { delete(e.inFlight, key) }() - errCh := make(chan error) + errCh := make(chan error, 1) go func() { errCh <- fn() }() t := time.NewTimer(e.maxWait) + var poisonedDone <-chan struct{} + if e.recoverOnTimeout { + poisonedDone = e.poisonedDone + } select { case err := <-errCh: t.Stop() - return err - case <-t.C: - buf := new(strings.Builder) - pprof.Lookup("goroutine").WriteTo(buf, 1) - e.logf("wgengine watchdog stacks:\n%s", buf.String()) - - // Collect the list of in-flight operations for debugging. - var ( - b []byte - now = time.Now() - ) - e.inFlightMu.Lock() - for k, t := range e.inFlight { - dur := now.Sub(t).Round(time.Millisecond) - b = fmt.Appendf(b, "in-flight[%d]: name=%s duration=%v start=%s\n", k.ctr, k.op, dur, t.Format(time.RFC3339Nano)) + if e.recoverOnTimeout && e.poisoned.Load() { + <-e.poisonedDone + return ErrWatchdogTimeout, true } - e.inFlightMu.Unlock() + return err, true + case <-poisonedDone: + t.Stop() + return ErrWatchdogTimeout, true + case <-t.C: + return e.watchdogTimeout(name), true + } +} - // Print everything as a single string to avoid log - // rate limits. - e.logf("wgengine watchdog in-flight:\n%s", b) +func (e *watchdogEngine) watchdogTimeout(name string) error { + firstTimeout := e.poison(name) + buf := new(strings.Builder) + pprof.Lookup("goroutine").WriteTo(buf, 1) + e.logf("wgengine watchdog stacks:\n%s", buf.String()) + + // Collect the list of in-flight operations for debugging. + var ( + b []byte + now = time.Now() + ) + e.inFlightMu.Lock() + for k, t := range e.inFlight { + dur := now.Sub(t).Round(time.Millisecond) + b = fmt.Appendf(b, "in-flight[%d]: name=%s duration=%v start=%s\n", k.ctr, k.op, dur, t.Format(time.RFC3339Nano)) + } + e.inFlightMu.Unlock() + + // Print everything as a single string to avoid log + // rate limits. + e.logf("wgengine watchdog in-flight:\n%s", b) + if firstTimeout && !e.recoverOnTimeout && e.fatalf != nil { e.fatalf("wgengine: watchdog timeout on %s", name) - return nil } + if e.recoverOnTimeout { + <-e.poisonedDone + return ErrWatchdogTimeout + } + return nil +} + +func (e *watchdogEngine) poison(operation string) bool { + if !e.poisoned.CompareAndSwap(false, true) { + return false + } + defer close(e.poisonedDone) + if e.recoverOnTimeout && e.timeoutCallback != nil { + e.timeoutCallback(operation) + } + return true } func (e *watchdogEngine) watchdog(name string, fn func()) { @@ -123,19 +228,42 @@ func (e *watchdogEngine) Reconfig(cfg *wgcfg.Config, routerCfg *router.Config, d return e.watchdogErr("Reconfig", func() error { return e.wrap.Reconfig(cfg, routerCfg, dnsCfg, debug) }) } func (e *watchdogEngine) GetFilter() *filter.Filter { + if e.poisoned.Load() { + return nil + } return e.wrap.GetFilter() } func (e *watchdogEngine) SetFilter(filt *filter.Filter) { e.watchdog("SetFilter", func() { e.wrap.SetFilter(filt) }) } func (e *watchdogEngine) SetStatusCallback(cb StatusCallback) { - e.watchdog("SetStatusCallback", func() { e.wrap.SetStatusCallback(cb) }) + if cb == nil { + e.watchdog("SetStatusCallback", func() { e.wrap.SetStatusCallback(nil) }) + return + } + e.watchdog("SetStatusCallback", func() { + e.wrap.SetStatusCallback(func(status *Status, err error) { + if !e.poisoned.Load() { + cb(status, err) + } + }) + }) } func (e *watchdogEngine) UpdateStatus(sb *ipnstate.StatusBuilder) { e.watchdog("UpdateStatus", func() { e.wrap.UpdateStatus(sb) }) } func (e *watchdogEngine) SetNetInfoCallback(cb NetInfoCallback) { - e.watchdog("SetNetInfoCallback", func() { e.wrap.SetNetInfoCallback(cb) }) + if cb == nil { + e.watchdog("SetNetInfoCallback", func() { e.wrap.SetNetInfoCallback(nil) }) + return + } + e.watchdog("SetNetInfoCallback", func() { + e.wrap.SetNetInfoCallback(func(netInfo *tailcfg.NetInfo) { + if !e.poisoned.Load() { + cb(netInfo) + } + }) + }) } func (e *watchdogEngine) RequestStatus() { e.watchdog("RequestStatus", func() { e.wrap.RequestStatus() }) @@ -150,16 +278,67 @@ func (e *watchdogEngine) SetNetworkMap(nm *netmap.NetworkMap) { e.watchdog("SetNetworkMap", func() { e.wrap.SetNetworkMap(nm) }) } func (e *watchdogEngine) AddNetworkMapCallback(callback NetworkMapCallback) func() { - var fn func() - e.watchdog("AddNetworkMapCallback", func() { fn = e.wrap.AddNetworkMapCallback(callback) }) - return func() { e.watchdog("RemoveNetworkMapCallback", fn) } + resultCh := make(chan func(), 1) + err, started := e.watchdogErrStarted("AddNetworkMapCallback", func() error { + resultCh <- e.wrap.AddNetworkMapCallback(func(networkMap *netmap.NetworkMap) { + if !e.poisoned.Load() { + callback(networkMap) + } + }) + return nil + }) + if err != nil { + if started { + go func() { + if remove := <-resultCh; remove != nil { + remove() + } + }() + } + return func() {} + } + remove := <-resultCh + if remove == nil { + return func() {} + } + if e.poisoned.Load() { + go remove() + return func() {} + } + var removeOnce sync.Once + return func() { + removeOnce.Do(func() { + _, started := e.watchdogErrStarted("RemoveNetworkMapCallback", func() error { + remove() + return nil + }) + if !started { + go remove() + } + }) + } } -func (e *watchdogEngine) DiscoPublicKey() (k key.DiscoPublic) { - e.watchdog("DiscoPublicKey", func() { k = e.wrap.DiscoPublicKey() }) +func (e *watchdogEngine) DiscoPublicKey() key.DiscoPublic { + k, ok := watchdogValue(e, "DiscoPublicKey", func() key.DiscoPublic { + return e.wrap.DiscoPublicKey() + }) + if !ok { + return key.DiscoPublic{} + } return k } func (e *watchdogEngine) Ping(ip netip.Addr, pingType tailcfg.PingType, cb func(*ipnstate.PingResult)) { - e.watchdog("Ping", func() { e.wrap.Ping(ip, pingType, cb) }) + if cb == nil { + e.watchdog("Ping", func() { e.wrap.Ping(ip, pingType, nil) }) + return + } + e.watchdog("Ping", func() { + e.wrap.Ping(ip, pingType, func(result *ipnstate.PingResult) { + if !e.poisoned.Load() { + cb(result) + } + }) + }) } func (e *watchdogEngine) RegisterIPPortIdentity(ipp netip.AddrPort, tsIP netip.Addr) { e.watchdog("RegisterIPPortIdentity", func() { e.wrap.RegisterIPPortIdentity(ipp, tsIP) }) @@ -167,22 +346,74 @@ func (e *watchdogEngine) RegisterIPPortIdentity(ipp netip.AddrPort, tsIP netip.A func (e *watchdogEngine) UnregisterIPPortIdentity(ipp netip.AddrPort) { e.watchdog("UnregisterIPPortIdentity", func() { e.wrap.UnregisterIPPortIdentity(ipp) }) } -func (e *watchdogEngine) WhoIsIPPort(ipp netip.AddrPort) (tsIP netip.Addr, ok bool) { - e.watchdog("UnregisterIPPortIdentity", func() { tsIP, ok = e.wrap.WhoIsIPPort(ipp) }) - return tsIP, ok +func (e *watchdogEngine) WhoIsIPPort(ipp netip.AddrPort) (netip.Addr, bool) { + result, ok := watchdogValue(e, "WhoIsIPPort", func() whoIsIPPortResult { + tsIP, ok := e.wrap.WhoIsIPPort(ipp) + return whoIsIPPortResult{tsIP: tsIP, ok: ok} + }) + if !ok { + return netip.Addr{}, false + } + return result.tsIP, result.ok } func (e *watchdogEngine) Close() { - e.watchdog("Close", e.wrap.Close) + if e.closeStarted.CompareAndSwap(false, true) { + go func() { + e.wrap.Close() + close(e.closeDone) + }() + } + if e.poisoned.Load() { + return + } + t := time.NewTimer(e.maxWait) + select { + case <-e.closeDone: + t.Stop() + case <-e.poisonedDone: + t.Stop() + case <-t.C: + _ = e.watchdogTimeout("Close") + } } -func (e *watchdogEngine) PeerForIP(ip netip.Addr) (ret PeerForIP, ok bool) { - e.watchdog("PeerForIP", func() { ret, ok = e.wrap.PeerForIP(ip) }) - return ret, ok +func (e *watchdogEngine) PeerForIP(ip netip.Addr) (PeerForIP, bool) { + result, ok := watchdogValue(e, "PeerForIP", func() peerForIPResult { + peer, ok := e.wrap.PeerForIP(ip) + return peerForIPResult{peer: peer, ok: ok} + }) + if !ok { + return PeerForIP{}, false + } + return result.peer, result.ok } func (e *watchdogEngine) Wait() { - e.wrap.Wait() + if e.poisoned.Load() { + return + } + if e.waitStarted.CompareAndSwap(false, true) { + go func() { + e.wrap.Wait() + close(e.wrappedWaitDone) + }() + } + select { + case <-e.wrappedWaitDone: + case <-e.poisonedDone: + } } func (e *watchdogEngine) InstallCaptureHook(cb capture.Callback) { - e.wrap.InstallCaptureHook(cb) + if e.poisoned.Load() { + return + } + if cb == nil { + e.wrap.InstallCaptureHook(nil) + return + } + e.wrap.InstallCaptureHook(func(path capture.Path, at time.Time, data []byte, meta packet.CaptureMeta) { + if !e.poisoned.Load() { + cb(path, at, data, meta) + } + }) } diff --git a/wgengine/watchdog_test.go b/wgengine/watchdog_test.go index da453606a203e..55459bd84a67d 100644 --- a/wgengine/watchdog_test.go +++ b/wgengine/watchdog_test.go @@ -4,11 +4,135 @@ package wgengine import ( + "net/netip" + "os" + "os/exec" "runtime" + "strings" + "sync/atomic" "testing" "time" + + "tailscale.com/envknob" + "tailscale.com/ipn/ipnstate" + "tailscale.com/net/dns" + "tailscale.com/tailcfg" + "tailscale.com/types/netmap" + "tailscale.com/wgengine/router" + "tailscale.com/wgengine/wgcfg" ) +const watchdogTestTimeout = time.Second + +type watchdogTestEngine struct { + Engine + + reconfigCalls atomic.Int32 + reconfigEntered chan<- struct{} + reconfigRelease <-chan struct{} + reconfigErr error + + closeCalls atomic.Int32 + closeEntered chan<- struct{} + closeRelease <-chan struct{} + + waitEntered chan<- struct{} + waitRelease <-chan struct{} + + peerEntered chan<- struct{} + peerRelease <-chan struct{} + peerResult PeerForIP + peerOK bool + + pingCallback chan<- func(*ipnstate.PingResult) + + addNetworkMapCallbackCalls atomic.Int32 + removeNetworkMapCallbackCalls atomic.Int32 + addNetworkMapCallbackEntered chan<- struct{} + addNetworkMapCallbackRelease <-chan struct{} + removeNetworkMapCallbackDone chan<- struct{} +} + +func (e *watchdogTestEngine) Reconfig(*wgcfg.Config, *router.Config, *dns.Config, *tailcfg.Debug) error { + e.reconfigCalls.Add(1) + if e.reconfigEntered != nil { + e.reconfigEntered <- struct{}{} + } + if e.reconfigRelease != nil { + <-e.reconfigRelease + } + return e.reconfigErr +} + +func (e *watchdogTestEngine) Close() { + e.closeCalls.Add(1) + if e.closeEntered != nil { + e.closeEntered <- struct{}{} + } + if e.closeRelease != nil { + <-e.closeRelease + } +} + +func (e *watchdogTestEngine) Wait() { + if e.waitEntered != nil { + e.waitEntered <- struct{}{} + } + if e.waitRelease != nil { + <-e.waitRelease + } +} + +func (e *watchdogTestEngine) PeerForIP(netip.Addr) (PeerForIP, bool) { + if e.peerEntered != nil { + e.peerEntered <- struct{}{} + } + if e.peerRelease != nil { + <-e.peerRelease + } + return e.peerResult, e.peerOK +} + +func (e *watchdogTestEngine) Ping(_ netip.Addr, _ tailcfg.PingType, callback func(*ipnstate.PingResult)) { + if e.pingCallback != nil { + e.pingCallback <- callback + } +} + +func (e *watchdogTestEngine) AddNetworkMapCallback(NetworkMapCallback) func() { + e.addNetworkMapCallbackCalls.Add(1) + if e.addNetworkMapCallbackEntered != nil { + e.addNetworkMapCallbackEntered <- struct{}{} + } + if e.addNetworkMapCallbackRelease != nil { + <-e.addNetworkMapCallbackRelease + } + return func() { + e.removeNetworkMapCallbackCalls.Add(1) + if e.removeNetworkMapCallbackDone != nil { + e.removeNetworkMapCallbackDone <- struct{}{} + } + } +} + +func newWatchdogTest(t *testing.T, e Engine, callback func(string)) *watchdogEngine { + t.Helper() + wrapped := NewWatchdogWithTimeoutCallback(e, callback) + wd := wrapped.(*watchdogEngine) + wd.maxWait = 20 * time.Millisecond + wd.logf = func(string, ...any) {} + return wd +} + +func waitWatchdogTest(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(watchdogTestTimeout): + t.Fatalf("timed out waiting for %s", what) + } +} + func TestWatchdog(t *testing.T) { t.Parallel() @@ -37,3 +161,506 @@ func TestWatchdog(t *testing.T) { e.Close() }) } + +func TestWatchdogDefaultTimeoutCallsFatal(t *testing.T) { + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wd := NewWatchdog(e).(*watchdogEngine) + wd.maxWait = 20 * time.Millisecond + wd.logf = func(string, ...any) {} + fatalCalled := make(chan struct{}, 1) + wd.fatalf = func(string, ...any) { + fatalCalled <- struct{}{} + } + + result := make(chan error, 1) + go func() { + result <- wd.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "default Reconfig") + waitWatchdogTest(t, fatalCalled, "default fatal hook") + close(reconfigRelease) + + select { + case err := <-result: + if err != nil { + t.Fatalf("default watchdog Reconfig error = %v, want nil", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for default Reconfig") + } +} + +func TestWatchdogDefaultTimeoutExitsProcess(t *testing.T) { + if os.Getenv("TS_WATCHDOG_FATAL_TEST_CHILD") == "1" { + envknob.Setenv("TS_DEBUG_DISABLE_WATCHDOG", "") + e := &watchdogTestEngine{ + reconfigRelease: make(chan struct{}), + } + wd := NewWatchdog(e).(*watchdogEngine) + wd.maxWait = 20 * time.Millisecond + wd.logf = func(string, ...any) {} + _ = wd.Reconfig(nil, nil, nil, nil) + return + } + + cmd := exec.Command(os.Args[0], "-test.run=^TestWatchdogDefaultTimeoutExitsProcess$") + cmd.Env = append(os.Environ(), "TS_WATCHDOG_FATAL_TEST_CHILD=1") + output, err := cmd.CombinedOutput() + exitErr, ok := err.(*exec.ExitError) + if !ok { + t.Fatalf("watchdog child error = %v, want process exit; output: %s", err, output) + } + if exitErr.ExitCode() != 1 { + t.Fatalf("watchdog child exit code = %d, want 1; output: %s", exitErr.ExitCode(), output) + } + if !strings.Contains(string(output), "wgengine: watchdog timeout on Reconfig") { + t.Fatalf("watchdog child output missing fatal timeout: %s", output) + } +} + +func TestWatchdogDisabledLeavesHungOperationRunning(t *testing.T) { + oldValue, hadValue := os.LookupEnv("TS_DEBUG_DISABLE_WATCHDOG") + envknob.Setenv("TS_DEBUG_DISABLE_WATCHDOG", "true") + t.Cleanup(func() { + if hadValue { + envknob.Setenv("TS_DEBUG_DISABLE_WATCHDOG", oldValue) + return + } + envknob.Setenv("TS_DEBUG_DISABLE_WATCHDOG", "") + }) + + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wrapped := NewWatchdog(e) + if wrapped != e { + t.Fatal("disabled watchdog wrapped the engine") + } + + result := make(chan error, 1) + go func() { + result <- wrapped.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "disabled Reconfig") + select { + case err := <-result: + t.Fatalf("hung operation unexpectedly returned: %v", err) + default: + } + + close(reconfigRelease) + select { + case err := <-result: + if err != nil { + t.Fatalf("disabled watchdog Reconfig error = %v, want nil", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for disabled Reconfig") + } +} + +func TestWatchdogTimeoutCallback(t *testing.T) { + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + callbackCalled := make(chan string, 1) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wd := newWatchdogTest(t, e, func(operation string) { + callbackCalled <- operation + }) + fatalCalled := make(chan struct{}, 1) + wd.fatalf = func(string, ...any) { + fatalCalled <- struct{}{} + } + + result := make(chan error, 1) + go func() { + result <- wd.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "callback Reconfig") + + select { + case err := <-result: + if err != ErrWatchdogTimeout { + t.Fatalf("callback Reconfig error = %v, want ErrWatchdogTimeout", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for callback Reconfig") + } + select { + case operation := <-callbackCalled: + if operation != "Reconfig" { + t.Fatalf("timeout callback operation = %q, want Reconfig", operation) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for timeout callback") + } + select { + case <-fatalCalled: + t.Fatal("callback watchdog called fatal hook") + default: + } + close(reconfigRelease) +} + +func TestWatchdogTimeoutCallbackOnceAndPoisonedOperations(t *testing.T) { + const operationCount = 3 + reconfigEntered := make(chan struct{}, operationCount) + reconfigRelease := make(chan struct{}) + callbackCalled := make(chan string, operationCount) + var callbackCount atomic.Int32 + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wd := newWatchdogTest(t, e, func(operation string) { + callbackCount.Add(1) + callbackCalled <- operation + }) + + results := make([]chan error, operationCount) + for i := range results { + results[i] = make(chan error, 1) + go func(result chan<- error) { + result <- wd.Reconfig(nil, nil, nil, nil) + }(results[i]) + } + for range operationCount { + waitWatchdogTest(t, reconfigEntered, "concurrent Reconfig") + } + select { + case <-callbackCalled: + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for first timeout callback") + } + + callsBefore := e.reconfigCalls.Load() + if err := wd.Reconfig(nil, nil, nil, nil); err != ErrWatchdogTimeout { + t.Fatalf("poisoned Reconfig error = %v, want ErrWatchdogTimeout", err) + } + if callsAfter := e.reconfigCalls.Load(); callsAfter != callsBefore { + t.Fatalf("poisoned Reconfig called underlying engine, calls = %d, want %d", callsAfter, callsBefore) + } + + close(reconfigRelease) + for i, result := range results { + select { + case err := <-result: + if err != ErrWatchdogTimeout { + t.Fatalf("concurrent Reconfig %d error = %v, want ErrWatchdogTimeout", i, err) + } + case <-time.After(watchdogTestTimeout): + t.Fatalf("timed out waiting for concurrent Reconfig %d", i) + } + } + if got := callbackCount.Load(); got != 1 { + t.Fatalf("timeout callback count = %d, want 1", got) + } +} + +func TestWatchdogPoisonedNetworkMapCallbackRemover(t *testing.T) { + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wd := newWatchdogTest(t, e, func(string) {}) + + result := make(chan error, 1) + go func() { + result <- wd.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "poisoning Reconfig") + select { + case err := <-result: + if err != ErrWatchdogTimeout { + t.Fatalf("poisoning Reconfig error = %v, want ErrWatchdogTimeout", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for poisoning Reconfig") + } + + addCalls := e.addNetworkMapCallbackCalls.Load() + remove := wd.AddNetworkMapCallback(nil) + if remove == nil { + t.Fatal("poisoned AddNetworkMapCallback returned nil remover") + } + remove() + remove() + if got := e.addNetworkMapCallbackCalls.Load(); got != addCalls { + t.Fatalf("poisoned AddNetworkMapCallback calls = %d, want %d", got, addCalls) + } + if got := e.removeNetworkMapCallbackCalls.Load(); got != 0 { + t.Fatalf("poisoned remover called underlying engine %d times, want 0", got) + } + close(reconfigRelease) +} + +func TestWatchdogPoisonedCloseReturnsPromptly(t *testing.T) { + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + closeEntered := make(chan struct{}, 1) + closeRelease := make(chan struct{}) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + closeEntered: closeEntered, + closeRelease: closeRelease, + } + wd := newWatchdogTest(t, e, func(string) {}) + + result := make(chan error, 1) + go func() { + result <- wd.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "poisoning Reconfig") + select { + case err := <-result: + if err != ErrWatchdogTimeout { + t.Fatalf("poisoning Reconfig error = %v, want ErrWatchdogTimeout", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for poisoning Reconfig") + } + close(reconfigRelease) + + closeDone := make(chan struct{}) + go func() { + wd.Close() + close(closeDone) + }() + waitWatchdogTest(t, closeDone, "poisoned Close return") + waitWatchdogTest(t, closeEntered, "underlying Close") + if got := e.closeCalls.Load(); got != 1 { + t.Fatalf("underlying Close calls = %d, want 1", got) + } + + waitDone := make(chan struct{}) + go func() { + wd.Wait() + close(waitDone) + }() + waitWatchdogTest(t, waitDone, "poisoned Wait return") + wd.Close() + if got := e.closeCalls.Load(); got != 1 { + t.Fatalf("underlying Close calls after second Close = %d, want 1", got) + } + close(closeRelease) +} + +func TestWatchdogTimedOutCloseAllowsSecondClose(t *testing.T) { + closeEntered := make(chan struct{}, 1) + closeRelease := make(chan struct{}) + e := &watchdogTestEngine{ + closeEntered: closeEntered, + closeRelease: closeRelease, + } + wd := newWatchdogTest(t, e, func(string) {}) + + firstDone := make(chan struct{}) + go func() { + wd.Close() + close(firstDone) + }() + waitWatchdogTest(t, closeEntered, "timed-out underlying Close") + waitWatchdogTest(t, firstDone, "timed-out Close return") + + secondDone := make(chan struct{}) + go func() { + wd.Close() + close(secondDone) + }() + waitWatchdogTest(t, secondDone, "second poisoned Close return") + close(closeRelease) +} + +func TestWatchdogTimedOutValueWorkerReturnsZero(t *testing.T) { + peerEntered := make(chan struct{}, 1) + peerRelease := make(chan struct{}) + e := &watchdogTestEngine{ + peerEntered: peerEntered, + peerRelease: peerRelease, + peerResult: PeerForIP{IsSelf: true}, + peerOK: true, + } + wd := newWatchdogTest(t, e, func(string) {}) + + result := make(chan struct { + peer PeerForIP + ok bool + }, 1) + go func() { + peer, ok := wd.PeerForIP(netip.Addr{}) + result <- struct { + peer PeerForIP + ok bool + }{peer: peer, ok: ok} + }() + waitWatchdogTest(t, peerEntered, "timed-out value operation") + + select { + case got := <-result: + if got.ok || got.peer.IsSelf { + t.Fatalf("timed-out value operation result = (%+v, %v), want zero", got.peer, got.ok) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for value operation") + } + close(peerRelease) +} + +func TestWatchdogInFlightWaitReturnsAfterPoison(t *testing.T) { + waitEntered := make(chan struct{}, 1) + waitRelease := make(chan struct{}) + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + e := &watchdogTestEngine{ + waitEntered: waitEntered, + waitRelease: waitRelease, + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + } + wd := newWatchdogTest(t, e, func(string) {}) + + waitDone := make(chan struct{}) + go func() { + wd.Wait() + close(waitDone) + }() + waitWatchdogTest(t, waitEntered, "underlying Wait") + + reconfigDone := make(chan error, 1) + go func() { + reconfigDone <- wd.Reconfig(nil, nil, nil, nil) + }() + waitWatchdogTest(t, reconfigEntered, "poisoning Reconfig") + select { + case err := <-reconfigDone: + if err != ErrWatchdogTimeout { + t.Fatalf("poisoning Reconfig error = %v, want ErrWatchdogTimeout", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for poisoning Reconfig") + } + waitWatchdogTest(t, waitDone, "in-flight Wait return") + close(reconfigRelease) + close(waitRelease) +} + +func TestWatchdogRemovesCallbackThatCompletesAfterTimeout(t *testing.T) { + addEntered := make(chan struct{}, 1) + addRelease := make(chan struct{}) + removeDone := make(chan struct{}, 1) + e := &watchdogTestEngine{ + addNetworkMapCallbackEntered: addEntered, + addNetworkMapCallbackRelease: addRelease, + removeNetworkMapCallbackDone: removeDone, + } + wd := newWatchdogTest(t, e, func(string) {}) + + addDone := make(chan func(), 1) + go func() { + addDone <- wd.AddNetworkMapCallback(func(*netmap.NetworkMap) {}) + }() + waitWatchdogTest(t, addEntered, "callback registration") + var remove func() + select { + case remove = <-addDone: + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for callback registration timeout") + } + remove() + close(addRelease) + waitWatchdogTest(t, removeDone, "late callback removal") +} + +func TestWatchdogPoisonWakesConcurrentOperations(t *testing.T) { + reconfigEntered := make(chan struct{}, 1) + reconfigRelease := make(chan struct{}) + peerEntered := make(chan struct{}, 1) + peerRelease := make(chan struct{}) + e := &watchdogTestEngine{ + reconfigEntered: reconfigEntered, + reconfigRelease: reconfigRelease, + peerEntered: peerEntered, + peerRelease: peerRelease, + peerResult: PeerForIP{IsSelf: true}, + peerOK: true, + } + wd := newWatchdogTest(t, e, func(string) {}) + wd.maxWait = time.Hour + + reconfigDone := make(chan error, 1) + go func() { + reconfigDone <- wd.Reconfig(nil, nil, nil, nil) + }() + peerDone := make(chan struct { + peer PeerForIP + ok bool + }, 1) + go func() { + peer, ok := wd.PeerForIP(netip.Addr{}) + peerDone <- struct { + peer PeerForIP + ok bool + }{peer: peer, ok: ok} + }() + waitWatchdogTest(t, reconfigEntered, "concurrent Reconfig") + waitWatchdogTest(t, peerEntered, "concurrent PeerForIP") + + if !wd.poison("test") { + t.Fatal("failed to poison healthy watchdog") + } + select { + case err := <-reconfigDone: + if err != ErrWatchdogTimeout { + t.Fatalf("concurrent Reconfig error = %v, want ErrWatchdogTimeout", err) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for concurrent Reconfig") + } + select { + case got := <-peerDone: + if got.ok || got.peer.IsSelf { + t.Fatalf("concurrent PeerForIP result = (%+v, %v), want zero", got.peer, got.ok) + } + case <-time.After(watchdogTestTimeout): + t.Fatal("timed out waiting for concurrent PeerForIP") + } + close(reconfigRelease) + close(peerRelease) +} + +func TestWatchdogQuarantinesCallbackAfterPoison(t *testing.T) { + pingCallback := make(chan func(*ipnstate.PingResult), 1) + callbackCalled := make(chan struct{}, 1) + e := &watchdogTestEngine{ + pingCallback: pingCallback, + } + wd := newWatchdogTest(t, e, func(string) {}) + + wd.Ping(netip.Addr{}, tailcfg.PingDisco, func(*ipnstate.PingResult) { + callbackCalled <- struct{}{} + }) + callback := <-pingCallback + if !wd.poison("test") { + t.Fatal("failed to poison healthy watchdog") + } + callback(&ipnstate.PingResult{}) + select { + case <-callbackCalled: + t.Fatal("callback ran after watchdog poison") + default: + } +}