fix: resolve concurrency bugs in eventbus and webtransport

p2p/host/eventbus/basic.go

@@ -3,7 +3,9 @@ package eventbus
 import (
 	"errors"
 	"fmt"
+	"log/slog"
 	"reflect"
+	"slices"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -25,6 +27,7 @@ type basicBus struct {
 	nodes         map[reflect.Type]*node
 	wildcard      *wildcardNode
 	metricsTracer MetricsTracer
+	log           *slog.Logger
 }
 
 var _ event.Bus = (*basicBus)(nil)
@@ -65,7 +68,8 @@ func (e *emitter) Close() error {
 func NewBus(opts ...Option) event.Bus {
 	bus := &basicBus{
 		nodes:    map[reflect.Type]*node{},
-		wildcard: &wildcardNode{},
+		wildcard: &wildcardNode{log: log},
+		log:      log,
 	}
 	for _, opt := range opts {
 		opt(bus)
@@ -78,7 +82,7 @@ func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node))
 
 	n, ok := b.nodes[typ]
 	if !ok {
-		n = newNode(typ, b.metricsTracer)
+		n = newNode(typ, b.metricsTracer, b.log)
 		b.nodes[typ] = n
 	}
 
@@ -334,8 +338,7 @@ type wildcardNode struct {
 	nSinks        atomic.Int32
 	sinks         []*namedSink
 	metricsTracer MetricsTracer
-
-	slowConsumerTimer *time.Timer
+	log           *slog.Logger
 }
 
 func (n *wildcardNode) addSink(sink *namedSink) {
@@ -350,22 +353,36 @@ func (n *wildcardNode) addSink(sink *namedSink) {
 }
 
 func (n *wildcardNode) removeSink(ch chan any) {
-	go func() {
-		// drain the event channel, will return when closed and drained.
-		// this is necessary to unblock publishes to this channel.
-		for range ch {
+	// Drain the event channel to unblock stalled emits, which hold the read
+	// lock; without this the Lock below would deadlock against them.
+	done := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		for {
+			select {
+			case <-ch:
+			case <-done:
+				// The write lock has been acquired: the sink is invisible to new
+				// emits and in-flight ones have completed, so only buffered
+				// events remain. Sweep them and exit.
+				for {
+					select {
+					case <-ch:
+					default:
+						return
+					}
+				}
+			}
 		}
-	}()
+	})
 	n.nSinks.Add(-1) // ok to do outside the lock
 	n.Lock()
-	for i := 0; i < len(n.sinks); i++ {
-		if n.sinks[i].ch == ch {
-			n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
-			n.sinks = n.sinks[:len(n.sinks)-1]
-			break
-		}
-	}
+	n.sinks = slices.DeleteFunc(n.sinks, func(s *namedSink) bool { return s.ch == ch })
 	n.Unlock()
+	// We could close ch itself here, which would also end the subscriber's
+	// Out() range like typed subs do.
+	close(done)
+	wg.Wait()
 }
 
 var wildcardType = reflect.TypeOf(event.WildcardSubscription)
@@ -385,12 +402,7 @@ func (n *wildcardNode) emit(evt any) {
 		select {
 		case sink.ch <- evt:
 		default:
-			slowConsumerTimer := emitAndLogError(n.slowConsumerTimer, wildcardType, evt, sink)
-			defer func() {
-				n.Lock()
-				n.slowConsumerTimer = slowConsumerTimer
-				n.Unlock()
-			}()
+			emitAndLogError(n.log, wildcardType, evt, sink)
 		}
 	}
 	n.RUnlock()
@@ -410,14 +422,14 @@ type node struct {
 
 	sinks         []*namedSink
 	metricsTracer MetricsTracer
-
-	slowConsumerTimer *time.Timer
+	log           *slog.Logger
 }
 
-func newNode(typ reflect.Type, metricsTracer MetricsTracer) *node {
+func newNode(typ reflect.Type, metricsTracer MetricsTracer, log *slog.Logger) *node {
 	return &node{
 		typ:           typ,
 		metricsTracer: metricsTracer,
+		log:           log,
 	}
 }
 
@@ -440,32 +452,24 @@ func (n *node) emit(evt any) {
 		select {
 		case sink.ch <- evt:
 		default:
-			n.slowConsumerTimer = emitAndLogError(n.slowConsumerTimer, n.typ, evt, sink)
+			emitAndLogError(n.log, n.typ, evt, sink)
 		}
 	}
 	n.lk.Unlock()
 }
 
-func emitAndLogError(timer *time.Timer, typ reflect.Type, evt any, sink *namedSink) *time.Timer {
+func emitAndLogError(log *slog.Logger, typ reflect.Type, evt any, sink *namedSink) {
 	// Slow consumer. Log a warning if stalled for the timeout
-	if timer == nil {
-		timer = time.NewTimer(slowConsumerWarningTimeout)
-	} else {
-		timer.Reset(slowConsumerWarningTimeout)
-	}
+	timer := time.NewTimer(slowConsumerWarningTimeout)
+	defer timer.Stop()
 
 	select {
 	case sink.ch <- evt:
-		if !timer.Stop() {
-			<-timer.C
-		}
 	case <-timer.C:
 		log.Warn("subscriber is a slow consumer. This can lead to libp2p stalling and hard to debug issues.", "subscriber_name", sink.name, "event_type", typ)
 		// Continue to stall since there's nothing else we can do.
 		sink.ch <- evt
 	}
-
-	return timer
 }
 
 func sendSubscriberMetrics(metricsTracer MetricsTracer, sink *namedSink) {

p2p/host/eventbus/basic_synctest_test.go

@@ -0,0 +1,194 @@
+//go:build go1.25
+
+package eventbus
+
+import (
+	"io"
+	"log/slog"
+	"slices"
+	"strings"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/event"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestWildcardSlowConsumerNoDeadlock checks that concurrent emitters to a full
+// wildcard queue all make progress once the queue drains, even after the
+// slow-consumer warning fires. Each emit must time its stall on its own timer:
+// a timer shared across concurrent emitters lets one emitter consume the single
+// tick and leaves the others blocked forever on the already-drained timer
+// channel, while holding the node's read lock.
+//
+// synctest gives us a fake clock, so the one-second warning timeout fires
+// instantly and deterministically instead of relying on real sleeps.
+func TestWildcardSlowConsumerNoDeadlock(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		const emitters = 3
+
+		// Capture log output to confirm the slow-consumer path actually ran.
+		logs := &mockLogger{}
+		bus := NewBus(withLogger(slog.New(slog.NewTextHandler(logs, nil))))
+		sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(1))
+		require.NoError(t, err)
+		defer sub.Close()
+
+		em, err := bus.Emitter(new(EventB))
+		require.NoError(t, err)
+		defer em.Close()
+
+		// drainOne frees a single buffer slot, letting one stalled emit proceed.
+		drainOne := func() { <-sub.Out() }
+
+		// Fill the single buffer slot so every later emit stalls on a full queue.
+		require.NoError(t, em.Emit(EventB(0)))
+
+		// Prime the slow-consumer path with one stalled emit, then release it
+		// before the timeout. The historical bug parked a single reusable timer
+		// on the node during this first stall; the concurrent emitters below then
+		// raced on it. The fix gives every emit its own timer, so this is now just
+		// a harmless warm-up, but it is what makes the deadlock reproducible.
+		primed := make(chan struct{})
+		go func() {
+			em.Emit(EventB(0))
+			close(primed)
+		}()
+		synctest.Wait() // primer is stalled in the slow path
+		drainOne()      // let the primer complete via a normal send
+		<-primed
+
+		// Concurrent emitters now all stall on the full queue and enter the slow
+		// path together.
+		done := make(chan struct{}, emitters)
+		for range emitters {
+			go func() {
+				em.Emit(EventB(0))
+				done <- struct{}{}
+			}()
+		}
+		synctest.Wait() // every emitter is blocked waiting on its timer
+
+		// Pass the warning timeout so every stalled emit's timer fires.
+		time.Sleep(slowConsumerWarningTimeout + time.Millisecond)
+		synctest.Wait() // every emitter has warned and is now blocked on its send
+
+		warnings := strings.Count(strings.Join(logs.Logs(), ""), "slow consumer")
+		require.Equal(t, emitters, warnings, "each stalled emitter should warn once")
+
+		// One drain per emitter; each frees a slot for exactly one stalled send.
+		for range emitters {
+			drainOne()
+		}
+
+		// Every emitter must return. A shared timer would leave one emitter stuck
+		// forever draining the already-consumed timer channel, which synctest
+		// reports as a deadlock.
+		for range emitters {
+			<-done
+		}
+	})
+}
+
+// TestWildcardSubCloseReleasesDrainGoroutine checks that closing a wildcard
+// subscription does not leak the drain goroutine started by removeSink. If the
+// drainer is still blocked when the bubble ends, synctest reports a deadlock.
+func TestWildcardSubCloseReleasesDrainGoroutine(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		bus := NewBus()
+		sub, err := bus.Subscribe(event.WildcardSubscription)
+		require.NoError(t, err)
+		require.NoError(t, sub.Close())
+	})
+}
+
+// TestWildcardCloseUnblocksStalledEmit covers the slow-consumer safety valve: an
+// emit stalled on a full wildcard subscriber, holding the node read lock, must
+// be released when the subscription is closed. wildcardNode.removeSink starts
+// draining the channel before it takes the write lock; were the order reversed,
+// Close would deadlock against the read lock the stalled emit still holds.
+func TestWildcardCloseUnblocksStalledEmit(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		// The stalled emit warns as part of this test; keep it out of the output.
+		bus := NewBus(withLogger(slog.New(slog.NewTextHandler(io.Discard, nil))))
+
+		// An unbuffered subscriber has nowhere to put the event, so the emit below
+		// stalls immediately without a fill step.
+		sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(0))
+		require.NoError(t, err)
+
+		em, err := bus.Emitter(new(EventB))
+		require.NoError(t, err)
+		defer em.Close()
+
+		emitReturned := make(chan struct{})
+		go func() {
+			em.Emit(EventB(0))
+			close(emitReturned)
+		}()
+
+		// Sleep past the warning timeout: the emit stalls, warns, and is now in
+		// the deepest stall state, a bare send with no timeout escape, still
+		// holding the node read lock.
+		time.Sleep(slowConsumerWarningTimeout + time.Millisecond)
+		synctest.Wait()
+
+		// Close must free the stalled emit. A regression here parks Close on
+		// wildcardNode's write lock, and a goroutine blocked on a sync.RWMutex is
+		// not durably blocked, so synctest cannot call it a deadlock: the package
+		// hangs until go test -timeout fires.
+		require.NoError(t, sub.Close())
+		<-emitReturned
+	})
+}
+
+func TestEmitLogsErrorOnStall(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ml := mockLogger{}
+		logger := slog.New(slog.NewTextHandler(&ml, nil))
+
+		bus1 := NewBus(withLogger(logger))
+		bus2 := NewBus(withLogger(logger))
+
+		eventSub, err := bus1.Subscribe(new(EventA))
+		require.NoError(t, err)
+
+		wildcardSub, err := bus2.Subscribe(event.WildcardSubscription)
+		require.NoError(t, err)
+
+		testCases := []event.Subscription{eventSub, wildcardSub}
+		eventBuses := []event.Bus{bus1, bus2}
+		names := []string{"typed sub", "wildcard sub"}
+
+		for i, sub := range testCases {
+			em, err := eventBuses[i].Emitter(new(EventA))
+			require.NoError(t, err)
+			defer em.Close()
+
+			// Overfill the subscriber's queue so the last emits hit the slow path.
+			go func() {
+				for range subSettingsDefault.buffer + 2 {
+					em.Emit(EventA{})
+				}
+			}()
+
+			// Pass the warning timeout on the fake clock, then let the stalled
+			// emitter warn.
+			time.Sleep(slowConsumerWarningTimeout + time.Millisecond)
+			synctest.Wait()
+
+			// Close the subscriber so the emitter can finish. Assert only once it
+			// has: a failed assertion unwinds the bubble's root goroutine, and a
+			// still-stalled emitter would then turn the failure into a timeout.
+			require.NoError(t, sub.Close())
+			synctest.Wait()
+
+			require.True(t, slices.ContainsFunc(ml.Logs(), func(l string) bool {
+				return strings.Contains(l, "slow consumer")
+			}), "expected to find slow consumer log for %s", names[i])
+			ml.Clear()
+		}
+	})
+}