Merge tag 'v1.90.9' into sunos-1.90

Release 1.90.9

Author: nshalman

VERSION.txt

@@ -1 +1 @@
-1.90.8
+1.90.9

util/eventbus/bus.go

@@ -120,7 +120,14 @@ func (b *Bus) Close() {
 }
 
 func (b *Bus) pump(ctx context.Context) {
-	var vals queue[PublishedEvent]
+	// Limit how many published events we can buffer in the PublishedEvent queue.
+	//
+	// Subscribers have unbounded DeliveredEvent queues (see tailscale/tailscale#18020),
+	// so this queue doesn't need to be unbounded. Keeping it bounded may also help
+	// catch cases where subscribers stop pumping events completely, such as due to a bug
+	// in [subscribeState.pump], [Subscriber.dispatch], or [SubscriberFunc.dispatch]).
+	const maxPublishedEvents = 16
+	vals := queue[PublishedEvent]{capacity: maxPublishedEvents}
 	acceptCh := func() chan PublishedEvent {
 		if vals.Full() {
 			return nil

util/eventbus/bus_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"log"
 	"regexp"
+	"sync"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -546,6 +547,105 @@ func TestRegression(t *testing.T) {
 	})
 }
 
+func TestPublishWithMutex(t *testing.T) {
+	testPublishWithMutex(t, 1024) // arbitrary large number of events
+}
+
+// testPublishWithMutex publishes the specified number of events,
+// acquiring and releasing a mutex around each publish and each
+// subscriber event receive.
+//
+// The test fails if it loses any events or times out due to a deadlock.
+// Unfortunately, a goroutine waiting on a mutex held by a durably blocked
+// goroutine is not itself considered durably blocked, so [synctest] cannot
+// detect this deadlock on its own.
+func testPublishWithMutex(t *testing.T, n int) {
+	synctest.Test(t, func(t *testing.T) {
+		b := eventbus.New()
+		defer b.Close()
+
+		c := b.Client("TestClient")
+
+		evts := make([]any, n)
+		for i := range evts {
+			evts[i] = EventA{Counter: i}
+		}
+		exp := expectEvents(t, evts...)
+
+		var mu sync.Mutex
+		eventbus.SubscribeFunc[EventA](c, func(e EventA) {
+			// Acquire the same mutex as the publisher.
+			mu.Lock()
+			mu.Unlock()
+
+			// Mark event as received, so we can check for lost events.
+			exp.Got(e)
+		})
+
+		p := eventbus.Publish[EventA](c)
+		go func() {
+			// Publish events, acquiring the mutex around each publish.
+			for i := range n {
+				mu.Lock()
+				p.Publish(EventA{Counter: i})
+				mu.Unlock()
+			}
+		}()
+
+		synctest.Wait()
+
+		if !exp.Empty() {
+			t.Errorf("unexpected extra events: %+v", exp.want)
+		}
+	})
+}
+
+func TestPublishFromSubscriber(t *testing.T) {
+	testPublishFromSubscriber(t, 1024) // arbitrary large number of events
+}
+
+// testPublishFromSubscriber publishes the specified number of EventA events.
+// Each EventA causes the subscriber to publish an EventB.
+// The test fails if it loses any events or if a deadlock occurs.
+func testPublishFromSubscriber(t *testing.T, n int) {
+	synctest.Test(t, func(t *testing.T) {
+		b := eventbus.New()
+		defer b.Close()
+
+		c := b.Client("TestClient")
+
+		// Ultimately we expect to receive n EventB events
+		// published as a result of receiving n EventA events.
+		evts := make([]any, n)
+		for i := range evts {
+			evts[i] = EventB{Counter: i}
+		}
+		exp := expectEvents(t, evts...)
+
+		pubA := eventbus.Publish[EventA](c)
+		pubB := eventbus.Publish[EventB](c)
+
+		eventbus.SubscribeFunc[EventA](c, func(e EventA) {
+			// Upon receiving EventA, publish EventB.
+			pubB.Publish(EventB{Counter: e.Counter})
+		})
+		eventbus.SubscribeFunc[EventB](c, func(e EventB) {
+			// Mark EventB as received.
+			exp.Got(e)
+		})
+
+		for i := range n {
+			pubA.Publish(EventA{Counter: i})
+		}
+
+		synctest.Wait()
+
+		if !exp.Empty() {
+			t.Errorf("unexpected extra events: %+v", exp.want)
+		}
+	})
+}
+
 type queueChecker struct {
 	t    *testing.T
 	want []any

util/eventbus/queue.go

@@ -7,18 +7,18 @@ import (
 	"slices"
 )
 
-const maxQueuedItems = 16
-
-// queue is an ordered queue of length up to maxQueuedItems.
+// queue is an ordered queue of length up to capacity,
+// if capacity is non-zero. Otherwise it is unbounded.
 type queue[T any] struct {
-	vals  []T
-	start int
+	vals     []T
+	start    int
+	capacity int // zero means unbounded
 }
 
 // canAppend reports whether a value can be appended to q.vals without
 // shifting values around.
 func (q *queue[T]) canAppend() bool {
-	return cap(q.vals) < maxQueuedItems || len(q.vals) < cap(q.vals)
+	return q.capacity == 0 || cap(q.vals) < q.capacity || len(q.vals) < cap(q.vals)
 }
 
 func (q *queue[T]) Full() bool {

wgengine/userspace.go

@@ -50,6 +50,7 @@ import (
 	"tailscale.com/util/checkchange"
 	"tailscale.com/util/clientmetric"
 	"tailscale.com/util/eventbus"
+	"tailscale.com/util/execqueue"
 	"tailscale.com/util/mak"
 	"tailscale.com/util/set"
 	"tailscale.com/util/testenv"
@@ -97,6 +98,8 @@ type userspaceEngine struct {
 	eventBus    *eventbus.Bus
 	eventClient *eventbus.Client
 
+	linkChangeQueue execqueue.ExecQueue
+
 	logf           logger.Logf
 	wgLogger       *wglog.Logger // a wireguard-go logging wrapper
 	reqCh          chan struct{}
@@ -543,7 +546,7 @@ func NewUserspaceEngine(logf logger.Logf, conf Config) (_ Engine, reterr error)
 		if f, ok := feature.HookProxyInvalidateCache.GetOk(); ok {
 			f()
 		}
-		e.linkChange(&cd)
+		e.linkChangeQueue.Add(func() { e.linkChange(&cd) })
 	})
 	e.eventClient = ec
 	e.logf("Engine created.")
@@ -1258,6 +1261,9 @@ func (e *userspaceEngine) RequestStatus() {
 
 func (e *userspaceEngine) Close() {
 	e.eventClient.Close()
+	// TODO(cmol): Should we wait for it too?
+	// Same question raised in appconnector.go.
+	e.linkChangeQueue.Shutdown()
 	e.mu.Lock()
 	if e.closing {
 		e.mu.Unlock()