appc,ipn/ipnlocal: add a required event bus to the AppConnector type (#17390)

Require the presence of the bus, but do not use it yet.  Check for required
fields and update tests and production use to plumb the necessary arguments.

Updates #15160
Updates #17192

Change-Id: I8cefd2fdb314ca9945317d3320bd5ea6a92e8dcb
Signed-off-by: M. J. Fromberger <fromberger@tailscale.com>

appc/appconnector.go

@@ -22,6 +22,7 @@ import (
 	"tailscale.com/types/views"
 	"tailscale.com/util/clientmetric"
 	"tailscale.com/util/dnsname"
+	"tailscale.com/util/eventbus"
 	"tailscale.com/util/execqueue"
 	"tailscale.com/util/slicesx"
 )
@@ -136,7 +137,9 @@ type RouteInfo struct {
 // routes not yet served by the AppConnector the local node configuration is
 // updated to advertise the new route.
 type AppConnector struct {
+	// These fields are immutable after initialization.
 	logf            logger.Logf
+	eventBus        *eventbus.Bus
 	routeAdvertiser RouteAdvertiser
 
 	// storeRoutesFunc will be called to persist routes if it is not nil.
@@ -168,6 +171,10 @@ type Config struct {
 	// It must be non-nil.
 	Logf logger.Logf
 
+	// EventBus receives events when the collection of routes maintained by the
+	// connector is updated. It must be non-nil.
+	EventBus *eventbus.Bus
+
 	// RouteAdvertiser allows the connector to update the set of advertised routes.
 	// It must be non-nil.
 	RouteAdvertiser RouteAdvertiser
@@ -183,8 +190,18 @@ type Config struct {
 
 // NewAppConnector creates a new AppConnector.
 func NewAppConnector(c Config) *AppConnector {
+	switch {
+	case c.Logf == nil:
+		panic("missing logger")
+	case c.EventBus == nil:
+		panic("missing event bus")
+	case c.RouteAdvertiser == nil:
+		panic("missing route advertiser")
+	}
+
 	ac := &AppConnector{
 		logf:            logger.WithPrefix(c.Logf, "appc: "),
+		eventBus:        c.EventBus,
 		routeAdvertiser: c.RouteAdvertiser,
 		storeRoutesFunc: c.StoreRoutesFunc,
 	}

appc/appconnector_test.go

@@ -4,7 +4,6 @@
 package appc
 
 import (
-	"context"
 	"net/netip"
 	"reflect"
 	"slices"
@@ -16,6 +15,7 @@ import (
 	"tailscale.com/appc/appctest"
 	"tailscale.com/tstest"
 	"tailscale.com/util/clientmetric"
+	"tailscale.com/util/eventbus/eventbustest"
 	"tailscale.com/util/mak"
 	"tailscale.com/util/must"
 	"tailscale.com/util/slicesx"
@@ -24,18 +24,20 @@ import (
 func fakeStoreRoutes(*RouteInfo) error { return nil }
 
 func TestUpdateDomains(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: &appctest.RouteCollector{},
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: &appctest.RouteCollector{}})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: &appctest.RouteCollector{}})
 		}
 		a.UpdateDomains([]string{"example.com"})
 
@@ -63,18 +65,20 @@ func TestUpdateDomains(t *testing.T) {
 }
 
 func TestUpdateRoutes(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{}, StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		a.updateDomains([]string{"*.example.com"})
 
@@ -116,19 +120,21 @@ func TestUpdateRoutes(t *testing.T) {
 }
 
 func TestUpdateRoutesUnadvertisesContainedRoutes(t *testing.T) {
-	ctx := context.Background()
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
 		rc := &appctest.RouteCollector{}
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		mak.Set(&a.domains, "example.com", []netip.Addr{netip.MustParseAddr("192.0.2.1")})
 		rc.SetRoutes([]netip.Prefix{netip.MustParsePrefix("192.0.2.1/32")})
@@ -143,24 +149,26 @@ func TestUpdateRoutesUnadvertisesContainedRoutes(t *testing.T) {
 }
 
 func TestDomainRoutes(t *testing.T) {
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
 		rc := &appctest.RouteCollector{}
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		a.updateDomains([]string{"example.com"})
 		if err := a.ObserveDNSResponse(dnsResponse("example.com.", "192.0.0.8")); err != nil {
 			t.Errorf("ObserveDNSResponse: %v", err)
 		}
-		a.Wait(context.Background())
+		a.Wait(t.Context())
 
 		want := map[string][]netip.Addr{
 			"example.com": {netip.MustParseAddr("192.0.0.8")},
@@ -173,19 +181,21 @@ func TestDomainRoutes(t *testing.T) {
 }
 
 func TestObserveDNSResponse(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 
 		// a has no domains configured, so it should not advertise any routes
@@ -267,19 +277,21 @@ func TestObserveDNSResponse(t *testing.T) {
 }
 
 func TestWildcardDomains(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 		var a *AppConnector
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 
 		a.updateDomains([]string{"*.example.com"})
@@ -422,8 +434,9 @@ func prefixes(in ...string) []netip.Prefix {
 }
 
 func TestUpdateRouteRouteRemoval(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 
 		assertRoutes := func(prefix string, routes, removedRoutes []netip.Prefix) {
@@ -439,12 +452,13 @@ func TestUpdateRouteRouteRemoval(t *testing.T) {
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		// nothing has yet been advertised
 		assertRoutes("appc init", []netip.Prefix{}, []netip.Prefix{})
@@ -472,8 +486,9 @@ func TestUpdateRouteRouteRemoval(t *testing.T) {
 }
 
 func TestUpdateDomainRouteRemoval(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 
 		assertRoutes := func(prefix string, routes, removedRoutes []netip.Prefix) {
@@ -489,12 +504,13 @@ func TestUpdateDomainRouteRemoval(t *testing.T) {
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		assertRoutes("appc init", []netip.Prefix{}, []netip.Prefix{})
 
@@ -532,8 +548,9 @@ func TestUpdateDomainRouteRemoval(t *testing.T) {
 }
 
 func TestUpdateWildcardRouteRemoval(t *testing.T) {
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	for _, shouldStore := range []bool{false, true} {
-		ctx := context.Background()
 		rc := &appctest.RouteCollector{}
 
 		assertRoutes := func(prefix string, routes, removedRoutes []netip.Prefix) {
@@ -549,12 +566,13 @@ func TestUpdateWildcardRouteRemoval(t *testing.T) {
 		if shouldStore {
 			a = NewAppConnector(Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = NewAppConnector(Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = NewAppConnector(Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		assertRoutes("appc init", []netip.Prefix{}, []netip.Prefix{})
 
@@ -691,10 +709,12 @@ func TestMetricBucketsAreSorted(t *testing.T) {
 // back into AppConnector via authReconfig. If everything is called
 // synchronously, this results in a deadlock on AppConnector.mu.
 func TestUpdateRoutesDeadlock(t *testing.T) {
-	ctx := context.Background()
+	ctx := t.Context()
+	bus := eventbustest.NewBus(t)
 	rc := &appctest.RouteCollector{}
 	a := NewAppConnector(Config{
 		Logf:            t.Logf,
+		EventBus:        bus,
 		RouteAdvertiser: rc,
 		RouteInfo:       &RouteInfo{},
 		StoreRoutesFunc: fakeStoreRoutes,

ipn/ipnlocal/local.go

@@ -4804,6 +4804,7 @@ func (b *LocalBackend) reconfigAppConnectorLocked(nm *netmap.NetworkMap, prefs i
 		}
 		b.appConnector = appc.NewAppConnector(appc.Config{
 			Logf:            b.logf,
+			EventBus:        b.sys.Bus.Get(),
 			RouteAdvertiser: b,
 			RouteInfo:       ri,
 			StoreRoutesFunc: storeFunc,

ipn/ipnlocal/local_test.go

@@ -2307,15 +2307,17 @@ func TestDNSConfigForNetmapForExitNodeConfigs(t *testing.T) {
 func TestOfferingAppConnector(t *testing.T) {
 	for _, shouldStore := range []bool{false, true} {
 		b := newTestBackend(t)
+		bus := b.sys.Bus.Get()
 		if b.OfferingAppConnector() {
 			t.Fatal("unexpected offering app connector")
 		}
+		rc := &appctest.RouteCollector{}
 		if shouldStore {
 			b.appConnector = appc.NewAppConnector(appc.Config{
-				Logf: t.Logf, RouteInfo: &appc.RouteInfo{}, StoreRoutesFunc: fakeStoreRoutes,
+				Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc, RouteInfo: &appc.RouteInfo{}, StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			b.appConnector = appc.NewAppConnector(appc.Config{Logf: t.Logf})
+			b.appConnector = appc.NewAppConnector(appc.Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		if !b.OfferingAppConnector() {
 			t.Fatal("unexpected not offering app connector")
@@ -2366,6 +2368,7 @@ func TestRouterAdvertiserIgnoresContainedRoutes(t *testing.T) {
 func TestObserveDNSResponse(t *testing.T) {
 	for _, shouldStore := range []bool{false, true} {
 		b := newTestBackend(t)
+		bus := b.sys.Bus.Get()
 
 		// ensure no error when no app connector is configured
 		if err := b.ObserveDNSResponse(dnsResponse("example.com.", "192.0.0.8")); err != nil {
@@ -2376,12 +2379,13 @@ func TestObserveDNSResponse(t *testing.T) {
 		if shouldStore {
 			b.appConnector = appc.NewAppConnector(appc.Config{
 				Logf:            t.Logf,
+				EventBus:        bus,
 				RouteAdvertiser: rc,
 				RouteInfo:       &appc.RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			b.appConnector = appc.NewAppConnector(appc.Config{Logf: t.Logf, RouteAdvertiser: rc})
+			b.appConnector = appc.NewAppConnector(appc.Config{Logf: t.Logf, EventBus: bus, RouteAdvertiser: rc})
 		}
 		b.appConnector.UpdateDomains([]string{"example.com"})
 		b.appConnector.Wait(context.Background())

ipn/ipnlocal/peerapi_test.go

@@ -259,12 +259,17 @@ func TestPeerAPIPrettyReplyCNAME(t *testing.T) {
 		if shouldStore {
 			a = appc.NewAppConnector(appc.Config{
 				Logf:            t.Logf,
+				EventBus:        sys.Bus.Get(),
 				RouteAdvertiser: &appctest.RouteCollector{},
 				RouteInfo:       &appc.RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = appc.NewAppConnector(appc.Config{Logf: t.Logf, RouteAdvertiser: &appctest.RouteCollector{}})
+			a = appc.NewAppConnector(appc.Config{
+				Logf:            t.Logf,
+				EventBus:        sys.Bus.Get(),
+				RouteAdvertiser: &appctest.RouteCollector{},
+			})
 		}
 		sys.Set(pm.Store())
 		sys.Set(eng)
@@ -339,12 +344,13 @@ func TestPeerAPIReplyToDNSQueriesAreObserved(t *testing.T) {
 		if shouldStore {
 			a = appc.NewAppConnector(appc.Config{
 				Logf:            t.Logf,
+				EventBus:        sys.Bus.Get(),
 				RouteAdvertiser: rc,
 				RouteInfo:       &appc.RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = appc.NewAppConnector(appc.Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = appc.NewAppConnector(appc.Config{Logf: t.Logf, EventBus: sys.Bus.Get(), RouteAdvertiser: rc})
 		}
 		sys.Set(pm.Store())
 		sys.Set(eng)
@@ -411,12 +417,13 @@ func TestPeerAPIReplyToDNSQueriesAreObservedWithCNAMEFlattening(t *testing.T) {
 		if shouldStore {
 			a = appc.NewAppConnector(appc.Config{
 				Logf:            t.Logf,
+				EventBus:        sys.Bus.Get(),
 				RouteAdvertiser: rc,
 				RouteInfo:       &appc.RouteInfo{},
 				StoreRoutesFunc: fakeStoreRoutes,
 			})
 		} else {
-			a = appc.NewAppConnector(appc.Config{Logf: t.Logf, RouteAdvertiser: rc})
+			a = appc.NewAppConnector(appc.Config{Logf: t.Logf, EventBus: sys.Bus.Get(), RouteAdvertiser: rc})
 		}
 		sys.Set(pm.Store())
 		sys.Set(eng)