wgengine: plumb an event bus into the userspace engine

Updates #15160

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

wgengine/bench/wg.go

@@ -46,7 +46,7 @@ func setupWGTest(b *testing.B, logf logger.Logf, traf *TrafficGen, a1, a2 netip.
 		logf: logger.WithPrefix(logf, "tun1: "),
 		traf: traf,
 	}
-	s1 := new(tsd.System)
+	s1 := tsd.NewSystemWithEventBus()
 	e1, err := wgengine.NewUserspaceEngine(l1, wgengine.Config{
 		Router:        router.NewFake(l1),
 		NetMon:        nil,
@@ -73,7 +73,7 @@ func setupWGTest(b *testing.B, logf logger.Logf, traf *TrafficGen, a1, a2 netip.
 		logf: logger.WithPrefix(logf, "tun2: "),
 		traf: traf,
 	}
-	s2 := new(tsd.System)
+	s2 := tsd.NewSystemWithEventBus()
 	e2, err := wgengine.NewUserspaceEngine(l2, wgengine.Config{
 		Router:        router.NewFake(l2),
 		NetMon:        nil,

wgengine/netstack/netstack_test.go

@@ -44,7 +44,7 @@ func TestInjectInboundLeak(t *testing.T) {
 			t.Logf(format, args...)
 		}
 	}
-	sys := new(tsd.System)
+	sys := tsd.NewSystemWithEventBus()
 	eng, err := wgengine.NewUserspaceEngine(logf, wgengine.Config{
 		Tun:           tunDev,
 		Dialer:        dialer,

wgengine/userspace.go

@@ -46,6 +46,7 @@ import (
 	"tailscale.com/types/views"
 	"tailscale.com/util/clientmetric"
 	"tailscale.com/util/deephash"
+	"tailscale.com/util/eventbus"
 	"tailscale.com/util/mak"
 	"tailscale.com/util/set"
 	"tailscale.com/util/testenv"
@@ -89,6 +90,10 @@ const statusPollInterval = 1 * time.Minute
 const networkLoggerUploadTimeout = 5 * time.Second
 
 type userspaceEngine struct {
+	// eventBus will eventually become required, but for now may be nil.
+	// TODO(creachadair): Enforce that this is non-nil at construction.
+	eventBus *eventbus.Bus
+
 	logf             logger.Logf
 	wgLogger         *wglog.Logger // a wireguard-go logging wrapper
 	reqCh            chan struct{}
@@ -227,6 +232,13 @@ type Config struct {
 	// DriveForLocal, if populated, will cause the engine to expose a Taildrive
 	// listener at 100.100.100.100:8080.
 	DriveForLocal drive.FileSystemForLocal
+
+	// EventBus, if non-nil, is used for event publication and subscription by
+	// the Engine and its subsystems.
+	//
+	// TODO(creachadair): As of 2025-03-19 this is optional, but is intended to
+	// become required non-nil.
+	EventBus *eventbus.Bus
 }
 
 // NewFakeUserspaceEngine returns a new userspace engine for testing.
@@ -255,6 +267,8 @@ func NewFakeUserspaceEngine(logf logger.Logf, opts ...any) (Engine, error) {
 			conf.HealthTracker = v
 		case *usermetric.Registry:
 			conf.Metrics = v
+		case *eventbus.Bus:
+			conf.EventBus = v
 		default:
 			return nil, fmt.Errorf("unknown option type %T", v)
 		}
@@ -323,6 +337,7 @@ func NewUserspaceEngine(logf logger.Logf, conf Config) (_ Engine, reterr error)
 	}
 
 	e := &userspaceEngine{
+		eventBus:       conf.EventBus,
 		timeNow:        mono.Now,
 		logf:           logf,
 		reqCh:          make(chan struct{}, 1),

wgengine/userspace_ext_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 func TestIsNetstack(t *testing.T) {
-	sys := new(tsd.System)
+	sys := tsd.NewSystemWithEventBus()
 	e, err := wgengine.NewUserspaceEngine(
 		tstest.WhileTestRunningLogger(t),
 		wgengine.Config{