diff --git a/p2p/host/eventbus/basic.go b/p2p/host/eventbus/basic.go
index a9db7b1e9a..2fd5387c02 100644
--- a/p2p/host/eventbus/basic.go
+++ b/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) {
diff --git a/p2p/host/eventbus/basic_synctest_test.go b/p2p/host/eventbus/basic_synctest_test.go
new file mode 100644
index 0000000000..0f31e8b699
--- /dev/null
+++ b/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()
+		}
+	})
+}
diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go
index b36c652102..269d40b051 100644
--- a/p2p/host/eventbus/basic_test.go
+++ b/p2p/host/eventbus/basic_test.go
@@ -3,9 +3,7 @@ package eventbus
 import (
 	"context"
 	"fmt"
-	"log/slog"
 	"reflect"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -160,62 +158,6 @@ func (m *mockLogger) Clear() {
 	m.logs = nil
 }
 
-func TestEmitLogsErrorOnStall(t *testing.T) {
-	oldLogger := log
-	defer func() {
-		log = oldLogger
-	}()
-	ml := mockLogger{}
-	log = slog.New(slog.NewTextHandler(&ml, nil))
-
-	bus1 := NewBus()
-	bus2 := NewBus()
-
-	eventSub, err := bus1.Subscribe(new(EventA))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	wildcardSub, err := bus2.Subscribe(event.WildcardSubscription)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	testCases := []event.Subscription{eventSub, wildcardSub}
-	eventBuses := []event.Bus{bus1, bus2}
-
-	for i, sub := range testCases {
-		bus := eventBuses[i]
-		em, err := bus.Emitter(new(EventA))
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer em.Close()
-
-		go func() {
-			for i := 0; i < subSettingsDefault.buffer+2; i++ {
-				em.Emit(EventA{})
-			}
-		}()
-
-		require.EventuallyWithT(t, func(collect *assert.CollectT) {
-			logs := ml.Logs()
-			found := false
-			for _, log := range logs {
-				if strings.Contains(log, "slow consumer") {
-					found = true
-					break
-				}
-			}
-			assert.True(collect, found, "expected to find slow consumer log")
-		}, 3*time.Second, 500*time.Millisecond)
-		ml.Clear()
-
-		// Close the subscriber so the worker can finish.
-		sub.Close()
-	}
-}
-
 func TestEmitOnClosed(t *testing.T) {
 	bus := NewBus()
 
diff --git a/p2p/host/eventbus/opts.go b/p2p/host/eventbus/opts.go
index f3e2a744ef..5943190b73 100644
--- a/p2p/host/eventbus/opts.go
+++ b/p2p/host/eventbus/opts.go
@@ -2,6 +2,7 @@ package eventbus
 
 import (
 	"fmt"
+	"log/slog"
 	"runtime"
 	"strings"
 	"sync/atomic"
@@ -77,3 +78,12 @@ func WithMetricsTracer(metricsTracer MetricsTracer) Option {
 		bus.wildcard.metricsTracer = metricsTracer
 	}
 }
+
+// withLogger sets the logger used by the bus. Defaults to the shared eventbus
+// logger.
+func withLogger(logger *slog.Logger) Option {
+	return func(bus *basicBus) {
+		bus.log = logger
+		bus.wildcard.log = logger
+	}
+}
diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go
index ede638fa85..0ba7875f36 100644
--- a/p2p/transport/webtransport/cert_manager.go
+++ b/p2p/transport/webtransport/cert_manager.go
@@ -6,6 +6,7 @@ import (
 	"crypto/tls"
 	"encoding/binary"
 	"fmt"
+	"slices"
 	"sync"
 	"time"
 
@@ -165,8 +166,17 @@ func (m *certManager) AddrComponent() ma.Multiaddr {
 	return m.addrComp
 }
 
+// SerializedCertHashes returns the multihash-encoded hashes of the currently
+// valid certificates. The caller owns the returned slice: cacheSerializedCertHashes
+// reuses the backing array across rotations, so handing out the manager's own
+// slice would let a caller observe it change underneath them.
+//
+// The copy is shallow. The hash byte slices are shared with the manager and
+// must be treated as read-only.
 func (m *certManager) SerializedCertHashes() [][]byte {
-	return m.serializedCertHashes
+	m.mx.RLock()
+	defer m.mx.RUnlock()
+	return slices.Clone(m.serializedCertHashes)
 }
 
 func (m *certManager) cacheSerializedCertHashes() error {
diff --git a/p2p/transport/webtransport/cert_manager_test.go b/p2p/transport/webtransport/cert_manager_test.go
index 9549a07dd4..aa86474ad9 100644
--- a/p2p/transport/webtransport/cert_manager_test.go
+++ b/p2p/transport/webtransport/cert_manager_test.go
@@ -4,6 +4,9 @@ import (
 	"crypto/sha256"
 	"crypto/tls"
 	"fmt"
+	"runtime"
+	"slices"
+	"sync"
 	"testing"
 	"testing/quick"
 	"time"
@@ -175,3 +178,96 @@ func TestGetCurrentBucketStartTimeIsWithinBounds(t *testing.T) {
 		return !bucketStart.After(start.Add(-clockSkewAllowance)) || bucketStart.Equal(start.Add(-clockSkewAllowance))
 	}, nil))
 }
+
+// TestSerializedCertHashesReturnsOuterCopy guards the copy in
+// SerializedCertHashes. cacheSerializedCertHashes truncates the hash slice and
+// re-appends into the same backing array across rotations, so returning the
+// manager's own slice would expose the caller to it changing underneath them.
+//
+// Only the outer slice is checked, matching what SerializedCertHashes promises.
+// Do not extend this to scribble over the hash bytes: they are shared with the
+// manager by design, so want aliases them and cannot be an oracle for their
+// contents. Such a test would zero the manager's live certificate hashes and
+// still report PASS.
+func TestSerializedCertHashesReturnsOuterCopy(t *testing.T) {
+	cl := clock.NewMock()
+	cl.Add(time.Hour * 24 * 365)
+	priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0)
+	require.NoError(t, err)
+	m, err := newCertManager(priv, cl)
+	require.NoError(t, err)
+	defer m.Close()
+
+	first := m.SerializedCertHashes()
+	require.NotEmpty(t, first)
+	want := slices.Clone(first)
+
+	// Scribble over the outer slots the way an owning caller may.
+	clear(first)
+
+	require.Equal(t, want, m.SerializedCertHashes(),
+		"caller must own the outer slice returned by SerializedCertHashes")
+}
+
+// TestSerializedCertHashesConcurrentWithRotation guards the read lock and the
+// copy in SerializedCertHashes. The background goroutine rewrites the hash slice
+// in place during certificate rotation, so a caller reading the slice at the same
+// time used to race with that write and could see a torn result. Run with -race
+// it fails on both an unsynchronized getter and one that hands out the manager's
+// own slice.
+func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) {
+	cl := clock.NewMock()
+	cl.Add(time.Hour * 24 * 365)
+	priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0)
+	require.NoError(t, err)
+	m, err := newCertManager(priv, cl)
+	require.NoError(t, err)
+	defer m.Close()
+
+	firstConf := m.GetConfig()
+
+	// Readers hammer the getters while certificates roll in the background. Run
+	// with -race: without the read lock the getter races with
+	// cacheSerializedCertHashes rewriting the slice during rotation, and without
+	// the copy the store below lands in the manager's own backing array. The
+	// sibling getters share the same lock, so read them here too.
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	for range 4 {
+		wg.Go(func() {
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+					// Write to a slot the caller owns. An indexed store is compiled
+					// to an instrumented write; clear() lowers to a bulk memclr in
+					// the runtime, which carries no instrumentation, so the race
+					// detector would never see it.
+					if h := m.SerializedCertHashes(); len(h) > 0 {
+						h[0] = nil
+					}
+					_ = m.GetConfig()
+					_ = m.AddrComponent()
+					// Yield: four spinning readers otherwise starve the rotation
+					// loop below, which costs seconds on a low-core CI runner.
+					runtime.Gosched()
+				}
+			}
+		})
+	}
+
+	// Advancing past a validity period forces the background goroutine to roll
+	// the config, which rewrites the shared serializedCertHashes slice.
+	for range 50 {
+		cl.Add(certValidity)
+		runtime.Gosched()
+	}
+
+	close(stop)
+	wg.Wait()
+
+	// Confirm rotation actually happened, so the reads above really did overlap
+	// with writes rather than racing nothing.
+	require.NotSame(t, firstConf, m.GetConfig(), "expected at least one certificate rotation")
+}