wgengine, ipn/ipnlocal, wgengine/magicsock: remove RequestStatus, eventbus-ify thingsbradfitz/getstatus

Updates #17900

Change-Id: Ia53a3f195a82256d8f915c8928d8a775f723259d
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>

2 files changed, 59 insertions, 87 deletions

diff --git a/ipn/ipnlocal/local.go b/ipn/ipnlocal/local.go
index f0a77531b..0b065e7cb 100644
--- a/ipn/ipnlocal/local.go
+++ b/ipn/ipnlocal/local.go
@@ -81,6 +81,7 @@ import (
 	"tailscale.com/types/persist"
 	"tailscale.com/types/preftype"
 	"tailscale.com/types/ptr"
+	"tailscale.com/types/topics"
 	"tailscale.com/types/views"
 	"tailscale.com/util/checkchange"
 	"tailscale.com/util/clientmetric"
@@ -287,7 +288,6 @@ type LocalBackend struct {
 	hostinfo          *tailcfg.Hostinfo      // TODO(nickkhyl): move to nodeBackend
 	nmExpiryTimer     tstime.TimerController // for updating netMap on node expiry; can be nil; TODO(nickkhyl): move to nodeBackend
 	activeLogin       string                 // last logged LoginName from netMap; TODO(nickkhyl): move to nodeBackend (or remove? it's in [ipn.LoginProfile]).
-	engineStatus      ipn.EngineStatus
 	endpoints         []tailcfg.Endpoint
 	blocked           bool
 	keyExpired        bool          // TODO(nickkhyl): move to nodeBackend
@@ -300,10 +300,11 @@ type LocalBackend struct {
 	peerAPIListeners  []*peerAPIListener
 	loginFlags        controlclient.LoginFlags
 	notifyWatchers    map[string]*watchSession // by session ID
-	lastStatusTime    time.Time                // status.AsOf value of the last processed status update
 	componentLogUntil map[string]componentLogState
 	currentUser       ipnauth.Actor
 
+	liveDERPs int // number of live DERP connections, per eventbus notification
+
 	// capForcedNetfilter is the netfilter that control instructs Linux clients
 	// to use, unless overridden locally.
 	capForcedNetfilter string // TODO(nickkhyl): move to nodeBackend
@@ -558,8 +559,6 @@ func NewLocalBackend(logf logger.Logf, logID logid.PublicID, sys *tsd.System, lo
 
 	b.setTCPPortsIntercepted(nil)
 
-	b.e.SetStatusCallback(b.setWgengineStatus)
-
 	b.prevIfState = netMon.InterfaceState()
 	// Call our linkChange code once with the current state.
 	// Following changes are triggered via the eventbus.
@@ -590,6 +589,8 @@ func NewLocalBackend(logf logger.Logf, logID logid.PublicID, sys *tsd.System, lo
 	ec := b.Sys().Bus.Get().Client("ipnlocal.LocalBackend")
 	b.eventClient = ec
 	eventbus.SubscribeFunc(ec, b.onClientVersion)
+	eventbus.SubscribeFunc(ec, b.onEndpointsChange)
+	eventbus.SubscribeFunc(ec, b.onDERPConnChange)
 	eventbus.SubscribeFunc(ec, func(au controlclient.AutoUpdate) {
 		b.onTailnetDefaultAutoUpdate(au.Value)
 	})
@@ -2251,65 +2252,25 @@ func (b *LocalBackend) resolveExitNodeIPLocked(prefs *ipn.Prefs) (prefsChanged b
 	return prefsChanged
 }
 
-// setWgengineStatus is the callback by the wireguard engine whenever it posts a new status.
-// This updates the endpoints both in the backend and in the control client.
-func (b *LocalBackend) setWgengineStatus(s *wgengine.Status, err error) {
-	if err != nil {
-		b.logf("wgengine status error: %v", err)
-		return
-	}
-	if s == nil {
-		b.logf("[unexpected] non-error wgengine update with status=nil: %v", s)
-		return
-	}
-
+func (b *LocalBackend) onEndpointsChange(eps topics.EndpointsChanged) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	// For now, only check this in the callback, but don't check it in setWgengineStatusLocked
-	if s.AsOf.Before(b.lastStatusTime) {
-		// Don't process a status update that is older than the one we have
-		// already processed. (corp#2579)
-		return
-	}
-	b.lastStatusTime = s.AsOf
-
-	b.setWgengineStatusLocked(s)
-}
-
-// setWgengineStatusLocked updates LocalBackend's view of the engine status and
-// updates the endpoints both in the backend and in the control client.
-//
-// Unlike setWgengineStatus it does not discard out-of-order updates, so
-// statuses sent here are always processed. This is useful for ensuring we don't
-// miss a "we shut down" status during backend shutdown even if other statuses
-// arrive out of order.
-//
-// TODO(zofrex): we should ensure updates actually do arrive in order and move
-// the out-of-order check into this function.
-//
-// b.mu must be held.
-func (b *LocalBackend) setWgengineStatusLocked(s *wgengine.Status) {
-	es := b.parseWgStatusLocked(s)
 	cc := b.cc
-
-	// TODO(zofrex): the only reason we even write this is to transition from
-	// "Starting" to "Running" in the call to state machine a few lines below
-	// this. Maybe we don't even need to store it at all.
-	b.engineStatus = es
-
-	needUpdateEndpoints := !slices.Equal(s.LocalAddrs, b.endpoints)
-	if needUpdateEndpoints {
-		b.endpoints = append([]tailcfg.Endpoint{}, s.LocalAddrs...)
+	if cc != nil {
+		cc.UpdateEndpoints(eps)
+		b.stateMachineLocked()
+		b.endpoints = append([]tailcfg.Endpoint{}, eps...)
 	}
+}
 
-	if cc != nil {
-		if needUpdateEndpoints {
-			cc.UpdateEndpoints(s.LocalAddrs)
-		}
+func (b *LocalBackend) onDERPConnChange(c topics.DERPConnChange) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.liveDERPs = c.LiveDERPs
+	if b.state == ipn.Starting {
 		b.stateMachineLocked()
 	}
-	b.sendLocked(ipn.Notify{Engine: &es})
 }
 
 // SetNotifyCallback sets the function to call when the backend has something to
@@ -3214,15 +3175,27 @@ func appendHealthActions(fn func(roNotify *ipn.Notify) (keepGoing bool)) func(*i
 	}
 }
 
-// pollRequestEngineStatus calls b.e.RequestStatus every 2 seconds until ctx
-// is done.
+// pollRequestEngineStatus calls b.e.RequestStatus every 2 seconds until ctx is
+// done.
+//
+// TODO(bradfitz): this is all too heavy and doesn't scale with large numbers of
+// clients. See tailscale/tailscale#1909, tailscale/tailscale#13392,
+// tailscale/tailscale#13392.
 func (b *LocalBackend) pollRequestEngineStatus(ctx context.Context) {
 	ticker, tickerChannel := b.clock.NewTicker(2 * time.Second)
 	defer ticker.Stop()
+	var last *wgengine.Status
 	for {
 		select {
 		case <-tickerChannel:
-			b.e.RequestStatus()
+			st := b.e.GetStatus()
+			if reflect.DeepEqual(last, st) {
+				continue
+			}
+			b.mu.Lock()
+			stBusForm := b.parseWgStatusLocked(st)
+			b.mu.Unlock()
+			b.send(ipn.Notify{Engine: &stBusForm})
 		case <-ctx.Done():
 			return
 		}
@@ -5660,8 +5633,6 @@ func (b *LocalBackend) enterStateLocked(newState ipn.State) {
 		}
 	case ipn.Starting, ipn.NeedsMachineAuth:
 		b.authReconfigLocked()
-		// Needed so that UpdateEndpoints can run
-		b.goTracker.Go(b.e.RequestStatus)
 	case ipn.Running:
 		if feature.CanSystemdStatus {
 			var addrStrs []string
@@ -5703,7 +5674,6 @@ func (b *LocalBackend) nextStateLocked() ipn.State {
 		netMap     = cn.NetMap()
 		state      = b.state
 		blocked    = b.blocked
-		st         = b.engineStatus
 		keyExpired = b.keyExpired
 
 		wantRunning = false
@@ -5754,7 +5724,7 @@ func (b *LocalBackend) nextStateLocked() ipn.State {
 		// (if we get here, we know MachineAuthorized == true)
 		return ipn.Starting
 	case state == ipn.Starting:
-		if st.NumLive > 0 || st.LiveDERPs > 0 {
+		if b.e.NumConfiguredPeers() > 0 || b.liveDERPs > 0 {
 			return ipn.Running
 		} else {
 			return state
@@ -5782,8 +5752,7 @@ func (b *LocalBackend) stateMachineLocked() {
 // b.mu must be held.
 func (b *LocalBackend) stopEngineAndWaitLocked() {
 	b.logf("stopEngineAndWait...")
-	st, _ := b.e.ResetAndStop() // TODO: what should we do if this returns an error?
-	b.setWgengineStatusLocked(st)
+	b.e.ResetAndStop()
 	b.logf("stopEngineAndWait: done.")
 }
 
diff --git a/ipn/ipnlocal/state_test.go b/ipn/ipnlocal/state_test.go
index 2197112b2..54fb9402b 100644
--- a/ipn/ipnlocal/state_test.go
+++ b/ipn/ipnlocal/state_test.go
@@ -39,6 +39,7 @@ import (
 	"tailscale.com/types/netmap"
 	"tailscale.com/types/persist"
 	"tailscale.com/types/preftype"
+	"tailscale.com/types/topics"
 	"tailscale.com/util/dnsname"
 	"tailscale.com/util/eventbus/eventbustest"
 	"tailscale.com/util/mak"
@@ -1004,7 +1005,7 @@ func runTestStateMachine(t *testing.T, seamless bool) {
 	}
 	notifies.expect(1)
 	// Fake a DERP connection.
-	b.setWgengineStatus(&wgengine.Status{DERPs: 1, AsOf: time.Now()}, nil)
+	b.onDERPConnChange(topics.DERPConnChange{RegionID: 1, Connected: true, LiveDERPs: 1})
 	{
 		nn := notifies.drain(1)
 		cc.assertCalls()
@@ -1144,11 +1145,11 @@ func TestWGEngineStatusRace(t *testing.T) {
 		wg.Add(1)
 		go func(i int) {
 			defer wg.Done()
-			n := 0
 			if i == 0 {
-				n = 1
+				b.onDERPConnChange(topics.DERPConnChange{RegionID: 1, Connected: true, LiveDERPs: 1})
+			} else {
+				b.onDERPConnChange(topics.DERPConnChange{RegionID: 1, Connected: false, LiveDERPs: 0})
 			}
-			b.setWgengineStatus(&wgengine.Status{AsOf: time.Now(), DERPs: n}, nil)
 		}(i)
 	}
 	wg.Wait()
@@ -1615,7 +1616,7 @@ func runTestSendPreservesAuthURL(t *testing.T, seamless bool) {
 	}})
 
 	t.Logf("Running")
-	b.setWgengineStatus(&wgengine.Status{AsOf: time.Now(), DERPs: 1}, nil)
+	b.onDERPConnChange(topics.DERPConnChange{RegionID: 1, Connected: true, LiveDERPs: 1})
 
 	t.Logf("Re-auth (StartLoginInteractive)")
 	b.StartLoginInteractive(t.Context())
@@ -1781,10 +1782,9 @@ type mockEngine struct {
 	cfg       *wgcfg.Config
 	routerCfg *router.Config
 	dnsCfg    *dns.Config
+	status    *wgengine.Status
 
 	filter, jailedFilter *filter.Filter
-
-	statusCb wgengine.StatusCallback
 }
 
 func newMockEngine() *mockEngine {
@@ -1805,6 +1805,24 @@ func (e *mockEngine) Reconfig(cfg *wgcfg.Config, routerCfg *router.Config, dnsCf
 	return nil
 }
 
+func (e *mockEngine) GetStatus() *wgengine.Status {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.status == nil {
+		return &wgengine.Status{}
+	}
+	return e.status
+}
+
+func (e *mockEngine) NumConfiguredPeers() int {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.status == nil {
+		return 0
+	}
+	return len(e.status.Peers)
+}
+
 func (e *mockEngine) Config() *wgcfg.Config {
 	e.mu.Lock()
 	defer e.mu.Unlock()
@@ -1851,27 +1869,12 @@ func (e *mockEngine) SetJailedFilter(f *filter.Filter) {
 	e.mu.Unlock()
 }
 
-func (e *mockEngine) SetStatusCallback(cb wgengine.StatusCallback) {
-	e.mu.Lock()
-	e.statusCb = cb
-	e.mu.Unlock()
-}
-
-func (e *mockEngine) RequestStatus() {
-	e.mu.Lock()
-	cb := e.statusCb
-	e.mu.Unlock()
-	if cb != nil {
-		cb(&wgengine.Status{AsOf: time.Now()}, nil)
-	}
-}
-
 func (e *mockEngine) ResetAndStop() (*wgengine.Status, error) {
 	err := e.Reconfig(&wgcfg.Config{}, &router.Config{}, &dns.Config{})
 	if err != nil {
 		return nil, err
 	}
-	return &wgengine.Status{AsOf: time.Now()}, nil
+	return &wgengine.Status{}, nil
 }
 
 func (e *mockEngine) PeerByKey(key.NodePublic) (_ wgint.Peer, ok bool) {